package callmemory import ( "context" "data-terminal-cloud/internal/identity" "data-terminal-cloud/internal/memoryidentity" "encoding/json" "errors" "github.com/jackc/pgx/v5" "strings" "time" ) // These records come directly from the native chat namespace, never the memoir mirror. type VikingRecord struct { ID string `json:"id"` Kind string `json:"kind"` Time int64 `json:"time_ms"` SessionID string `json:"session_id"` Content string `json:"content"` MemoryInfo map[string]json.RawMessage `json:"memory_info"` Digest string `json:"digest"` } type VikingConsoleProvider interface { ChatMemories(context.Context, string, string, string, int, int64, int64) ([]VikingRecord, bool, error) DeleteChatMemory(context.Context, string, string) error } func (s *Service) ConfigureVikingConsole(p VikingConsoleProvider) { s.vikingConsole = p } type VikingQuery struct { ElderID string `json:"elder_id"` Kind string `json:"kind"` Query string `json:"query"` Limit int `json:"limit"` Start int64 `json:"start_time_ms"` End int64 `json:"end_time_ms"` } type VikingDelete struct { ElderID string `json:"elder_id"` OperationID string `json:"operation_id"` Kind string `json:"kind"` ID string `json:"id"` Digest string `json:"digest"` Time int64 `json:"time_ms"` Reason string `json:"reason"` } func (s *Service) vikingRequest(ctx context.Context, p identity.Principal, op string, raw []byte) (any, error) { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() if op == "viking/query" { var in VikingQuery if e := StrictJSON(raw, &in); e != nil { return nil, e } if in.Kind != "event" && in.Kind != "profile" || len([]rune(in.Query)) > 500 || in.Start < 0 || in.End < 0 || in.End > 0 && in.Start > in.End || in.Kind == "profile" && (in.Start != 0 || in.End != 0) { return nil, fault(400, "INVALID_VIKING_QUERY") } if in.Limit == 0 { in.Limit = 100 } if in.Limit < 1 || in.Limit > 1000 { return nil, fault(400, "INVALID_VIKING_QUERY") } a, e := ResolveAccess(ctx, s.pool, p, in.ElderID, false) if e != nil { return nil, e } if !a.Sensitive { return nil, fault(403, "VIKING_SENSITIVE_ACCESS_REQUIRED") } if s.vikingConsole == nil { return nil, fault(503, "VIKING_NOT_CONFIGURED") } items, limited, e := s.vikingConsole.ChatMemories(ctx, memoryidentity.UserID(a.Tenant, a.Subject), in.Kind, strings.TrimSpace(in.Query), in.Limit, in.Start, in.End) if e != nil { return nil, fault(503, "VIKING_QUERY_FAILED") } return map[string]any{"items": items, "can_write": a.Write, "limited": limited, "limit": in.Limit, "source": "VIKING_NATIVE_CHAT"}, nil } var in VikingDelete if e := StrictJSON(raw, &in); e != nil { return nil, e } in.Reason = strings.TrimSpace(in.Reason) if !validUUID(in.OperationID) || !short(in.ID) || (in.Kind != "event" && in.Kind != "profile") || len(in.Digest) != 64 || in.Time < 0 || in.Reason == "" || len([]rune(in.Reason)) > 500 { return nil, fault(400, "INVALID_VIKING_DELETE") } conn, e := s.pool.Acquire(ctx) if e != nil { return nil, e } defer conn.Release() a, e := ResolveAccess(ctx, conn, p, in.ElderID, true) if e != nil { return nil, e } if !a.Sensitive { return nil, fault(403, "VIKING_SENSITIVE_ACCESS_REQUIRED") } if s.vikingConsole == nil { return nil, fault(503, "VIKING_NOT_CONFIGURED") } // A session lock spans the durable request receipt and the external deletion. lock := a.Subject + ":" + in.Kind + ":" + in.ID var locked bool if e = conn.QueryRow(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,93))`, lock).Scan(&locked); e != nil { return nil, e } if !locked { return nil, fault(409, "VIKING_DELETE_BUSY") } defer func() { c, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if _, e := conn.Exec(c, `SELECT pg_advisory_unlock(hashtextextended($1,93))`, lock); e != nil { _ = conn.Hijack().Close(c) } }() fingerprint := hash(jsonBytes(struct { Subject, Actor string Input VikingDelete }{a.Subject, a.User, in})) var receipt []byte e = conn.QueryRow(ctx, `SELECT detail FROM audit_event WHERE id=$1::uuid AND action='viking.memory.delete'`, in.OperationID).Scan(&receipt) existing := e == nil if e != nil && !errors.Is(e, pgx.ErrNoRows) { return nil, e } if existing { var prior struct { Fingerprint string `json:"fingerprint"` State string `json:"state"` } if json.Unmarshal(receipt, &prior) != nil || prior.Fingerprint != fingerprint { return nil, fault(409, "VIKING_OPERATION_CONFLICT") } if prior.State == "DELETED" { return map[string]any{"deleted": true, "operation_id": in.OperationID}, nil } } user := memoryidentity.UserID(a.Tenant, a.Subject) // Verify by ID across the scope. A changed timestamp must not look deleted. start, end := int64(0), int64(0) items, limited, e := s.vikingConsole.ChatMemories(ctx, user, in.Kind, "", 5000, start, end) if e != nil { return nil, fault(503, "VIKING_QUERY_FAILED") } var current *VikingRecord for i := range items { if items[i].ID == in.ID { current = &items[i] break } } if current == nil && (!existing || limited) { return nil, fault(404, "VIKING_RECORD_NOT_FOUND") } if current != nil && current.Digest != in.Digest { return nil, fault(409, "VIKING_RECORD_CHANGED") } if !existing { // Fail closed if recording the request fails. The receipt contains no memory body. detail := jsonBytes(map[string]any{"fingerprint": fingerprint, "state": "REQUESTED", "subject_id": a.Subject, "record_id": in.ID, "kind": in.Kind, "digest": in.Digest, "reason": in.Reason, "actor_id": a.User}) tag, e := conn.Exec(ctx, `INSERT INTO audit_event(id,tenant_id,actor_type,action,resource_type,resource_id,request_id,purpose,result,detail) VALUES($1::uuid,$2::uuid,'user','viking.memory.delete','subject',$3::uuid,$1,'用户删除聊天记忆','failure',$4::jsonb) ON CONFLICT(id) DO NOTHING`, in.OperationID, a.Tenant, a.Subject, detail) if e != nil { return nil, e } if tag.RowsAffected() != 1 { return nil, fault(409, "VIKING_OPERATION_CONFLICT") } } finish := func(state, result string) error { _, e := conn.Exec(ctx, `UPDATE audit_event SET result=$2,detail=jsonb_set(detail,'{state}',to_jsonb($3::text)) WHERE id=$1::uuid AND action='viking.memory.delete'`, in.OperationID, result, state) return e } if current != nil { if e = s.vikingConsole.DeleteChatMemory(ctx, in.Kind, in.ID); e != nil { _ = finish("UNKNOWN", "failure") return nil, fault(503, "VIKING_DELETE_UNCONFIRMED") } items, limited, e = s.vikingConsole.ChatMemories(ctx, user, in.Kind, "", 5000, start, end) if e != nil { _ = finish("UNKNOWN", "failure") return nil, fault(503, "VIKING_DELETE_UNCONFIRMED") } for _, item := range items { if item.ID == in.ID { _ = finish("UNKNOWN", "failure") return nil, fault(503, "VIKING_DELETE_UNCONFIRMED") } } if limited { _ = finish("UNKNOWN", "failure") return nil, fault(503, "VIKING_DELETE_UNCONFIRMED") } } if e = finish("DELETED", "success"); e != nil { return nil, e } return map[string]any{"deleted": true, "operation_id": in.OperationID}, nil }