DEV-000061 checkpoint interaction chat foundation

adapters/postgres/interaction_repository.go

@@ -0,0 +1,66 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"git.cin.su/m/Assistant/contracts/interaction"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type InteractionRepository struct{ pool *pgxpool.Pool }
+
+func OpenInteraction(ctx context.Context, c ContinuityConfig) (*InteractionRepository, error) {
+	p, e := pgxpool.New(ctx, fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s", c.Host, c.Port, c.Database, c.User, c.Password))
+	return &InteractionRepository{p}, e
+}
+func (r *InteractionRepository) Close() { r.pool.Close() }
+
+func (r *InteractionRepository) CreateConversation(ctx context.Context, c interaction.Conversation) error {
+	_, err := r.pool.Exec(ctx, `INSERT INTO interaction.conversations(conversation_id,actor_id,principal_id,channel,state,created_at_unix_millis,updated_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(conversation_id) DO NOTHING`, c.ID, c.ActorID, c.PrincipalID, c.Channel, c.State, c.CreatedAtUnixMillis, c.UpdatedAtUnixMillis)
+	return err
+}
+
+func (r *InteractionRepository) AppendUtterance(ctx context.Context, c interaction.Conversation, u interaction.HumanUtterance, intent interaction.OriginalHumanIntent, relations []interaction.InteractionRelation) error {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(ctx)
+	if _, err = tx.Exec(ctx, `INSERT INTO interaction.conversations(conversation_id,actor_id,principal_id,channel,state,created_at_unix_millis,updated_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(conversation_id) DO UPDATE SET updated_at_unix_millis=EXCLUDED.updated_at_unix_millis`, c.ID, c.ActorID, c.PrincipalID, c.Channel, c.State, c.CreatedAtUnixMillis, c.UpdatedAtUnixMillis); err != nil {
+		return err
+	}
+	if _, err = tx.Exec(ctx, `INSERT INTO interaction.utterances(utterance_id,conversation_id,actor_id,principal_id,channel,received_at_unix_millis,correlation_id,content,content_ref,speech_act,interpretation_state,corrects_utterance_id) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)`, u.ID, u.ConversationID, u.ActorID, u.PrincipalID, u.Channel, u.ReceivedAtUnixMillis, u.CorrelationID, u.Content, u.ContentRef, u.SpeechAct, u.InterpretationState, u.CorrectsUtteranceID); err != nil {
+		return err
+	}
+	if _, err = tx.Exec(ctx, `INSERT INTO interaction.original_human_intents(intent_id,utterance_id,recorded_at_unix_millis,text) VALUES($1,$2,$3,$4)`, intent.ID, intent.UtteranceID, intent.RecordedAtUnixMillis, intent.Text); err != nil {
+		return err
+	}
+	for _, relation := range relations {
+		if _, err = tx.Exec(ctx, `INSERT INTO interaction.relations(relation_id,from_utterance_id,to_utterance_id,kind,recorded_at_unix_millis) VALUES($1,$2,$3,$4,$5)`, relation.ID, relation.FromUtteranceID, relation.ToUtteranceID, relation.Kind, relation.RecordedAtUnixMillis); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
+}
+func (r *InteractionRepository) Conversation(ctx context.Context, id interaction.ConversationID) (interaction.Conversation, error) {
+	var c interaction.Conversation
+	e := r.pool.QueryRow(ctx, `SELECT conversation_id,actor_id,principal_id,channel,state,created_at_unix_millis,updated_at_unix_millis FROM interaction.conversations WHERE conversation_id=$1`, id).Scan(&c.ID, &c.ActorID, &c.PrincipalID, &c.Channel, &c.State, &c.CreatedAtUnixMillis, &c.UpdatedAtUnixMillis)
+	return c, e
+}
+func (r *InteractionRepository) Utterances(ctx context.Context, id interaction.ConversationID) ([]interaction.HumanUtterance, error) {
+	rows, e := r.pool.Query(ctx, `SELECT utterance_id,conversation_id,actor_id,principal_id,channel,received_at_unix_millis,correlation_id,content,content_ref,speech_act,interpretation_state,corrects_utterance_id FROM interaction.utterances WHERE conversation_id=$1 ORDER BY received_at_unix_millis,utterance_id`, id)
+	if e != nil {
+		return nil, e
+	}
+	defer rows.Close()
+	var out []interaction.HumanUtterance
+	for rows.Next() {
+		var u interaction.HumanUtterance
+		if e = rows.Scan(&u.ID, &u.ConversationID, &u.ActorID, &u.PrincipalID, &u.Channel, &u.ReceivedAtUnixMillis, &u.CorrelationID, &u.Content, &u.ContentRef, &u.SpeechAct, &u.InterpretationState, &u.CorrectsUtteranceID); e != nil {
+			return nil, e
+		}
+		out = append(out, u)
+	}
+	return out, rows.Err()
+}