107 lines
6.7 KiB
Go
107 lines
6.7 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
|
|
"git.cin.su/m/Assistant/contracts/identity"
|
|
"git.cin.su/m/Assistant/contracts/situation"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type SituationRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func OpenSituation(ctx context.Context, config ContinuityConfig) (*SituationRepository, error) {
|
|
pool, err := openPool(ctx, config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &SituationRepository{pool: pool}, nil
|
|
}
|
|
|
|
func (r *SituationRepository) SaveContextScope(ctx context.Context, scope situation.ContextScope) error {
|
|
if err := scope.Validate(); err != nil {
|
|
return err
|
|
}
|
|
bindings, err := json.Marshal(scope.Bindings)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = r.pool.Exec(ctx, `INSERT INTO situation.context_scopes(context_scope_id,principal_id,revision,bindings,recorded_at_unix_millis) VALUES($1,$2,$3,$4,$5) ON CONFLICT(context_scope_id) DO UPDATE SET revision=EXCLUDED.revision, bindings=EXCLUDED.bindings, recorded_at_unix_millis=EXCLUDED.recorded_at_unix_millis`, scope.ID, scope.PrincipalID, scope.Revision, bindings, scope.RecordedAt)
|
|
return err
|
|
}
|
|
|
|
func (r *SituationRepository) SaveFocusFrame(ctx context.Context, frame situation.FocusFrame) error {
|
|
if err := frame.Validate(); err != nil {
|
|
return err
|
|
}
|
|
ref, err := json.Marshal(frame.ContextScopeRef)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = r.pool.Exec(ctx, `INSERT INTO situation.focus_frames(focus_frame_id,principal_id,actor_id,revision,authority_scope_ref,context_scope_ref,goal_ref,task_ref,source_utterance_ref,provenance,confidence_millis,state,parent_frame_ref,activated_at_unix_millis,suspended_at_unix_millis,expired_at_unix_millis,recorded_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) ON CONFLICT(focus_frame_id) DO UPDATE SET revision=EXCLUDED.revision, context_scope_ref=EXCLUDED.context_scope_ref, state=EXCLUDED.state, suspended_at_unix_millis=EXCLUDED.suspended_at_unix_millis, expired_at_unix_millis=EXCLUDED.expired_at_unix_millis, recorded_at_unix_millis=EXCLUDED.recorded_at_unix_millis`, frame.ID, frame.PrincipalID, frame.ActorID, frame.Revision, frame.AuthorityScopeRef, ref, frame.GoalRef, frame.TaskRef, frame.SourceUtteranceRef, frame.Provenance, frame.ConfidenceMillis, frame.State, frame.ParentFrameRef, frame.ActivatedAt, frame.SuspendedAt, frame.ExpiredAt, frame.RecordedAt)
|
|
return err
|
|
}
|
|
|
|
func (r *SituationRepository) AppendFocusTransition(ctx context.Context, transition situation.FocusTransition) error {
|
|
if err := transition.Validate(); err != nil {
|
|
return err
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO situation.focus_transitions(transition_id,focus_frame_id,from_revision,to_revision,operation,cause_ref,recorded_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7)`, transition.TransitionID, transition.FrameID, transition.FromRevision, transition.ToRevision, transition.Operation, transition.CauseRef, transition.RecordedAt)
|
|
return err
|
|
}
|
|
func (r *SituationRepository) Close() { r.pool.Close() }
|
|
func (r *SituationRepository) Save(ctx context.Context, binding situation.FocusBinding) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO situation.focus_bindings (focus_binding_id, actor_id, session_id, workspace_id, project_id, goal_id, scope_id, active) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT (focus_binding_id) DO UPDATE SET active=EXCLUDED.active`, binding.ID, binding.ActorID, binding.SessionID, binding.Workspace, binding.ProjectID, binding.GoalID, binding.ScopeID, binding.Active)
|
|
return err
|
|
}
|
|
func (r *SituationRepository) ByRuntime(ctx context.Context, runtime situation.RuntimeBinding) ([]situation.FocusBinding, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT focus_binding_id, actor_id, session_id, workspace_id, project_id, goal_id, scope_id, active FROM situation.focus_bindings WHERE session_id=$1 AND workspace_id=$2`, runtime.SessionID, runtime.Workspace)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var bindings []situation.FocusBinding
|
|
for rows.Next() {
|
|
var binding situation.FocusBinding
|
|
if err := rows.Scan(&binding.ID, &binding.ActorID, &binding.SessionID, &binding.Workspace, &binding.ProjectID, &binding.GoalID, &binding.ScopeID, &binding.Active); err != nil {
|
|
return nil, err
|
|
}
|
|
bindings = append(bindings, binding)
|
|
}
|
|
return bindings, rows.Err()
|
|
}
|
|
|
|
func (r *SituationRepository) ListProjects(ctx context.Context) ([]situation.ProjectDescriptor, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT p.project_id, p.display_name, p.recorded_at_unix_millis, COALESCE(array_agg(a.alias ORDER BY a.alias) FILTER (WHERE a.alias IS NOT NULL), '{}') FROM situation.projects p LEFT JOIN situation.project_aliases a ON a.project_id=p.project_id GROUP BY p.project_id, p.display_name, p.recorded_at_unix_millis ORDER BY p.display_name`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var projects []situation.ProjectDescriptor
|
|
for rows.Next() {
|
|
var project situation.ProjectDescriptor
|
|
var aliases []string
|
|
if err := rows.Scan(&project.ID, &project.DisplayName, &project.RecordedAtUnixMillis, &aliases); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, alias := range aliases {
|
|
project.Aliases = append(project.Aliases, situation.ProjectAlias(strings.ToLower(alias)))
|
|
}
|
|
projects = append(projects, project)
|
|
}
|
|
return projects, rows.Err()
|
|
}
|
|
|
|
func (r *SituationRepository) ActiveProjectFocus(ctx context.Context, actorID identity.ActorID, principalID identity.PrincipalID) (situation.ActiveProjectFocus, error) {
|
|
var focus situation.ActiveProjectFocus
|
|
err := r.pool.QueryRow(ctx, `SELECT actor_id,principal_id,project_id,project_display_name,provenance,confidence_millis,source_utterance_id,updated_at_unix_millis FROM situation.active_project_focus WHERE actor_id=$1 AND principal_id=$2`, actorID, principalID).Scan(&focus.ActorID, &focus.PrincipalID, &focus.ProjectID, &focus.ProjectDisplayName, &focus.Provenance, &focus.ConfidenceMillis, &focus.SourceUtteranceID, &focus.UpdatedAtUnixMillis)
|
|
return focus, err
|
|
}
|
|
|
|
func (r *SituationRepository) SaveActiveProjectFocus(ctx context.Context, focus situation.ActiveProjectFocus) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO situation.active_project_focus(actor_id,principal_id,project_id,project_display_name,provenance,confidence_millis,source_utterance_id,updated_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(actor_id,principal_id) DO UPDATE SET project_id=EXCLUDED.project_id, project_display_name=EXCLUDED.project_display_name, provenance=EXCLUDED.provenance, confidence_millis=EXCLUDED.confidence_millis, source_utterance_id=EXCLUDED.source_utterance_id, updated_at_unix_millis=EXCLUDED.updated_at_unix_millis`, focus.ActorID, focus.PrincipalID, focus.ProjectID, focus.ProjectDisplayName, focus.Provenance, focus.ConfidenceMillis, focus.SourceUtteranceID, focus.UpdatedAtUnixMillis)
|
|
return err
|
|
}
|