package vikingmemory import ( "context" "data-terminal-cloud/internal/callmemory" "data-terminal-cloud/internal/outbox" "encoding/json" "errors" "fmt" "github.com/jackc/pgx/v5/pgxpool" "strconv" "strings" ) const ProjectionAssistantID = "platform_companion_v2" type ProjectionAdapter struct { pool *pgxpool.Pool client *Client } type projectionSchema struct { EventType string `json:"event_type"` ProfileType string `json:"profile_type"` IDField string `json:"memory_id_field"` VersionField string `json:"record_version_field"` OperationField string `json:"operation_key_field"` EventContent string `json:"event_content_field"` ProfileContent string `json:"profile_content_field"` ProfileGroups bool `json:"profile_groups_verified"` } func NewProjectionAdapter(pool *pgxpool.Pool, client *Client) *ProjectionAdapter { return &ProjectionAdapter{pool, client} } func (a *ProjectionAdapter) schema(ctx context.Context) (projectionSchema, error) { var p projectionSchema var raw []byte e := a.pool.QueryRow(ctx, `SELECT projection_schema FROM call_provider_profile WHERE published AND capabilities->>'H08'='true' AND COALESCE(evidence->>'H08','')<>'' ORDER BY created_at DESC LIMIT 1`).Scan(&raw) if e != nil { return p, &callmemory.Error{Status: 409, Code: "PROJECTION_SCHEMA_UNVERIFIED"} } if e = json.Unmarshal(raw, &p); e != nil { return p, e } if p.EventType == "" || p.IDField == "" || p.VersionField == "" || p.OperationField == "" || p.EventContent == "" { return p, &callmemory.Error{Status: 409, Code: "PROJECTION_SCHEMA_UNVERIFIED"} } return p, nil } // Project uses provider CRUD APIs, never AddSession. An uncertain add is only // reconciled by its persistent operation key; absence in a query is not permission // to create again. A full 5000-item result cannot prove absence and is blocked. func (a *ProjectionAdapter) Project(ctx context.Context, r callmemory.ProjectionRecord, priorKind, priorID string, reconcileOnly bool) (callmemory.ProjectionResult, error) { out := callmemory.ProjectionResult{} p, e := a.schema(ctx) if e != nil { return out, e } kind := "event" rule := p.EventType field := p.EventContent if r.Kind == "PROFILE" { if p.ProfileType == "" || p.ProfileContent == "" || !p.ProfileGroups { return out, &callmemory.Error{Status: 409, Code: "PROFILE_PROJECTION_UNVERIFIED"} } kind = "profile" rule = p.ProfileType field = p.ProfileContent } if priorID != "" && priorKind != "" && priorKind != kind { return out, &callmemory.Error{Status: 409, Code: "PROJECTION_KIND_CHANGE_REQUIRES_CLEANUP"} } user := UserID(r.Tenant, r.Subject) operation := r.ID + ":" + strconv.FormatInt(r.Version, 10) serve := r.Status == "ACTIVE" && r.Policy == "ALLOWED" && (r.Kind != "OPEN_ITEM" || r.ItemStatus == "OPEN") items, complete, e := a.list(ctx, kind, rule, user) if e != nil { return out, e } matches := []Item{} for _, item := range items { var id string _ = json.Unmarshal(item.MemoryInfo[p.IDField], &id) if id == r.ID { matches = append(matches, item) } } if len(matches) > 1 { return out, &callmemory.Error{Status: 409, Code: "DUPLICATE_PROVIDER_PROJECTION"} } if len(matches) == 1 { priorID = matches[0].ID out.ID = priorID out.Kind = kind var version int64 _ = json.Unmarshal(matches[0].MemoryInfo[p.VersionField], &version) var content string _ = json.Unmarshal(matches[0].MemoryInfo[field], &content) if serve && version == r.Version && content == r.Content { out.Visible = true return out, nil } } if !serve { if reconcileOnly && priorID == "" && len(matches) == 0 { return out, errors.New("uncertain create must be reconciled before proving deletion") } if len(matches) == 0 && complete { out.Kind = kind out.ID = priorID out.Visible = true return out, nil } if priorID == "" { return out, &callmemory.Error{Status: 409, Code: "PROVIDER_ABSENCE_NOT_PROVEN"} } e = a.client.call(ctx, "/api/memory/"+kind+"/delete", map[string]any{kind + "_id": priorID}, nil) if e != nil { return out, projectionError(e) } out.ID = priorID out.Kind = kind return out, nil } if reconcileOnly && priorID == "" { return out, errors.New("provider add outcome remains unknown") } if !complete && priorID == "" { return out, &callmemory.Error{Status: 409, Code: "PROVIDER_SCOPE_TOO_LARGE"} } info := map[string]any{p.IDField: r.ID, p.VersionField: r.Version, p.OperationField: operation, field: r.Content} payload := map[string]any{"memory_info": info} path := "/api/memory/" + kind + "/update" if priorID != "" { payload[kind+"_id"] = priorID } else { path = "/api/memory/" + kind + "/add" payload[kind+"_type"] = rule payload["user_id"] = user payload["assistant_id"] = ProjectionAssistantID if kind == "profile" { payload["group_id"] = r.ID payload["is_upsert"] = false } else { payload["update_profiles"] = []string{} } } var data map[string]json.RawMessage e = a.client.call(ctx, path, payload, &data) if e != nil { return out, projectionError(e) } if priorID == "" { _ = json.Unmarshal(data[kind+"_id"], &priorID) if strings.TrimSpace(priorID) == "" { return out, errors.New("provider acknowledgement has no object identity") } } out.Kind = kind out.ID = priorID out.Visible = false return out, nil } func (a *ProjectionAdapter) list(ctx context.Context, kind, rule, user string) ([]Item, bool, error) { var data struct { Items []Item `json:"result_list"` } e := a.client.call(ctx, "/api/memory/"+kind+"/search", map[string]any{"filter": map[string]any{"user_id": []string{user}, "assistant_id": []string{ProjectionAssistantID}, "memory_type": []string{rule}}, "limit": 5000}, &data) if e != nil { return nil, false, projectionError(e) } for _, item := range data.Items { if item.ID == "" || !contains(item.UserID, user) || !contains(item.AssistantID, ProjectionAssistantID) { return nil, false, &callmemory.Error{Status: 409, Code: "PROVIDER_SCOPE_VIOLATION"} } } return data.Items, len(data.Items) < 5000, nil } func projectionError(e error) error { if outbox.IsRetryable(e) { return e } return &callmemory.Error{Status: 409, Code: "PROVIDER_SCHEMA_OR_AUTH_REJECTED", Message: fmt.Sprintf("Provider rejected projection: %T", e)} }