feat: establish v0.70 platform milestone

2026-08-29 21:56:51 +08:00
parent d97bbf7cf5
commit e652516147
133 changed files with 12907 additions and 801 deletions
@@ -0,0 +1,116 @@
+package providermemory
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"data-terminal-cloud/internal/outbox"
+)
+
+type Memory struct {
+	TenantID       string `json:"tenant_id"`
+	HouseholdID    string `json:"household_id"`
+	SubjectID      string `json:"subject_id"`
+	MemoryID       string `json:"memory_id"`
+	MemoryVersion  string `json:"memory_version_id"`
+	VersionNo      int64  `json:"version_no"`
+	Content        string `json:"content"`
+	ContentDigest  string `json:"content_digest"`
+	IdempotencyKey string `json:"idempotency_key"`
+}
+
+type Client interface {
+	Upsert(context.Context, Memory) (string, error)
+	Delete(context.Context, string, string) error
+}
+
+type HTTPClient struct {
+	baseURL string
+	apiKey  string
+	client  *http.Client
+}
+
+func NewHTTPClient(baseURL, apiKey string, timeout time.Duration) (*HTTPClient, error) {
+	parsed, err := url.Parse(strings.TrimRight(strings.TrimSpace(baseURL), "/"))
+	if err != nil || parsed.Scheme == "" || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") {
+		return nil, errors.New("provider memory base URL must be an absolute HTTP(S) URL")
+	}
+	if strings.TrimSpace(apiKey) == "" || timeout <= 0 {
+		return nil, errors.New("provider memory API key and positive timeout are required")
+	}
+	return &HTTPClient{baseURL: parsed.String(), apiKey: strings.TrimSpace(apiKey), client: &http.Client{Timeout: timeout}}, nil
+}
+
+func (c *HTTPClient) Upsert(ctx context.Context, memory Memory) (string, error) {
+	payload, err := json.Marshal(memory)
+	if err != nil {
+		return "", err
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/v1/memories:upsert", bytes.NewReader(payload))
+	if err != nil {
+		return "", err
+	}
+	request.Header.Set("Authorization", "Bearer "+c.apiKey)
+	request.Header.Set("Content-Type", "application/json")
+	request.Header.Set("Idempotency-Key", memory.IdempotencyKey)
+	response, err := c.client.Do(request)
+	if err != nil {
+		return "", outbox.Retryable(fmt.Errorf("call provider memory upsert: %w", err))
+	}
+	defer response.Body.Close()
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return "", providerError("upsert", response)
+	}
+	var result struct {
+		ProviderRef string `json:"provider_ref"`
+	}
+	if err := json.NewDecoder(io.LimitReader(response.Body, 64*1024)).Decode(&result); err != nil {
+		return "", outbox.Retryable(errors.New("provider memory upsert returned invalid JSON"))
+	}
+	result.ProviderRef = strings.TrimSpace(result.ProviderRef)
+	if result.ProviderRef == "" || len(result.ProviderRef) > 300 {
+		return "", outbox.Retryable(errors.New("provider memory upsert returned an invalid reference"))
+	}
+	return result.ProviderRef, nil
+}
+
+func (c *HTTPClient) Delete(ctx context.Context, providerRef, idempotencyKey string) error {
+	providerRef = strings.TrimSpace(providerRef)
+	if providerRef == "" {
+		return nil
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.baseURL+"/v1/memories/"+url.PathEscape(providerRef), nil)
+	if err != nil {
+		return err
+	}
+	request.Header.Set("Authorization", "Bearer "+c.apiKey)
+	request.Header.Set("Idempotency-Key", idempotencyKey)
+	response, err := c.client.Do(request)
+	if err != nil {
+		return outbox.Retryable(fmt.Errorf("call provider memory delete: %w", err))
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusNotFound || response.StatusCode == http.StatusGone {
+		return nil
+	}
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return providerError("delete", response)
+	}
+	return nil
+}
+
+func providerError(action string, response *http.Response) error {
+	err := fmt.Errorf("provider memory %s returned HTTP %d", action, response.StatusCode)
+	if response.StatusCode == http.StatusTooManyRequests || response.StatusCode >= 500 {
+		return outbox.Retryable(err)
+	}
+	return err
+}
@@ -0,0 +1,55 @@
+package providermemory
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestHTTPClientUpsertAndDelete(t *testing.T) {
+	requests := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		if r.Header.Get("Authorization") != "Bearer test-key" || r.Header.Get("Idempotency-Key") == "" {
+			t.Fatalf("missing provider authentication or idempotency header")
+		}
+		switch {
+		case r.Method == http.MethodPost && r.URL.Path == "/v1/memories:upsert":
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"provider_ref":"provider-memory-1"}`))
+		case r.Method == http.MethodDelete && r.URL.Path == "/v1/memories/provider-memory-1":
+			w.WriteHeader(http.StatusNoContent)
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer server.Close()
+	client, err := NewHTTPClient(server.URL, "test-key", time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	providerRef, err := client.Upsert(context.Background(), Memory{TenantID: "tenant", HouseholdID: "household", SubjectID: "subject", MemoryID: "memory", MemoryVersion: "version", VersionNo: 1, Content: "content", ContentDigest: "digest", IdempotencyKey: "upsert-1"})
+	if err != nil || providerRef != "provider-memory-1" {
+		t.Fatalf("providerRef=%q err=%v", providerRef, err)
+	}
+	if err := client.Delete(context.Background(), providerRef, "delete-1"); err != nil {
+		t.Fatal(err)
+	}
+	if requests != 2 {
+		t.Fatalf("requests=%d", requests)
+	}
+}
+
+func TestHTTPClientTreatsMissingDeleteAsSuccess(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { http.NotFound(w, nil) }))
+	defer server.Close()
+	client, err := NewHTTPClient(server.URL, "test-key", time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := client.Delete(context.Background(), "missing", "delete-missing"); err != nil {
+		t.Fatal(err)
+	}
+}
@@ -0,0 +1,213 @@
+package providermemory
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"data-terminal-cloud/internal/audit"
+	"data-terminal-cloud/internal/database"
+	"data-terminal-cloud/internal/outbox"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+var ErrMemoryChangedDuringProviderSync = errors.New("memory changed during provider synchronization")
+
+type Processor struct {
+	pool   *pgxpool.Pool
+	client Client
+	audit  *audit.Repository
+}
+
+type providerSnapshot struct {
+	tenantInternalID string
+	tenantID         string
+	householdID      string
+	subjectID        string
+	memoryID         string
+	memoryVersionID  string
+	content          string
+	contentDigest    string
+	providerRef      string
+	status           string
+	currentVersionID string
+	sensitivity      string
+	versionNo        int64
+	consentEffective bool
+}
+
+func NewProcessor(pool *pgxpool.Pool, client Client) (*Processor, error) {
+	if pool == nil || client == nil {
+		return nil, errors.New("provider memory processor requires database and client")
+	}
+	return &Processor{pool: pool, client: client, audit: audit.NewRepository()}, nil
+}
+
+func (p *Processor) HandleSync(ctx context.Context, event outbox.Event) error {
+	var memoryVersionID string
+	if err := json.Unmarshal(event.Payload, &struct {
+		MemoryVersionID *string `json:"memory_version_id"`
+	}{MemoryVersionID: &memoryVersionID}); err != nil || memoryVersionID == "" {
+		return errors.New("provider memory sync event requires memory_version_id")
+	}
+	before, err := p.loadSnapshot(ctx, p.pool, memoryVersionID, false)
+	if err != nil {
+		return err
+	}
+	if !providerEligible(before, memoryVersionID) {
+		if before.providerRef != "" {
+			if err := p.client.Delete(ctx, before.providerRef, "provider-memory-stale:"+memoryVersionID); err != nil {
+				return p.markFailed(ctx, memoryVersionID, err)
+			}
+		}
+		_, err := p.pool.Exec(ctx, `UPDATE memory_index_entry SET status='STALE',provider_ref=NULL,last_attempt_at=now(),last_error='memory is not currently eligible',updated_at=now() WHERE memory_version_id=$1::uuid AND index_type='provider'`, memoryVersionID)
+		return err
+	}
+	if err := p.deleteSupersededRefs(ctx, before.memoryID, memoryVersionID); err != nil {
+		return p.markFailed(ctx, memoryVersionID, err)
+	}
+	providerRef, err := p.client.Upsert(ctx, Memory{
+		TenantID: before.tenantID, HouseholdID: before.householdID, SubjectID: before.subjectID,
+		MemoryID: before.memoryID, MemoryVersion: before.memoryVersionID, VersionNo: before.versionNo,
+		Content: before.content, ContentDigest: before.contentDigest, IdempotencyKey: "provider-memory-upsert:" + memoryVersionID,
+	})
+	if err != nil {
+		return p.markFailed(ctx, memoryVersionID, err)
+	}
+
+	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(context.Background())
+	current, err := p.loadSnapshot(ctx, tx, memoryVersionID, true)
+	if err != nil {
+		return err
+	}
+	if !providerEligible(current, memoryVersionID) || current.contentDigest != before.contentDigest {
+		_ = p.client.Delete(ctx, providerRef, "provider-memory-rollback:"+memoryVersionID)
+		return outbox.Retryable(ErrMemoryChangedDuringProviderSync)
+	}
+	if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET status='ACTIVE',provider_ref=$2,indexed_at=now(),last_attempt_at=now(),last_error=NULL,updated_at=now() WHERE memory_version_id=$1::uuid AND index_type='provider'`, memoryVersionID, providerRef); err != nil {
+		return err
+	}
+	if err := p.audit.Record(ctx, tx, audit.Event{TenantID: &current.tenantInternalID, ActorType: "system", Action: "memory.provider_sync.complete", ResourceType: "memory", ResourceID: &current.memoryID, Result: "success", Detail: map[string]any{"memory_version_id": current.memoryVersionID}}); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+func (p *Processor) HandleDelete(ctx context.Context, event outbox.Event) error {
+	rows, err := p.pool.Query(ctx, `SELECT memory_version_id::text,COALESCE(provider_ref,'') FROM memory_index_entry WHERE memory_id=$1::uuid AND index_type='provider' AND status<>'DELETED' ORDER BY created_at,id`, event.AggregateID)
+	if err != nil {
+		return err
+	}
+	type reference struct{ versionID, providerRef string }
+	refs := make([]reference, 0)
+	for rows.Next() {
+		var item reference
+		if err := rows.Scan(&item.versionID, &item.providerRef); err != nil {
+			rows.Close()
+			return err
+		}
+		refs = append(refs, item)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	for _, item := range refs {
+		if item.providerRef != "" {
+			if err := p.client.Delete(ctx, item.providerRef, "provider-memory-delete:"+item.versionID); err != nil {
+				return p.markFailed(ctx, item.versionID, err)
+			}
+		}
+	}
+	return database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
+		var tenantID, memoryPublicID string
+		err := tx.QueryRow(ctx, `SELECT tenant_id::text,public_id::text FROM curated_memory WHERE id=$1::uuid FOR UPDATE`, event.AggregateID).Scan(&tenantID, &memoryPublicID)
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,provider_ref=NULL,status='DELETED',indexed_at=NULL,last_attempt_at=now(),last_error=NULL,updated_at=now() WHERE memory_id=$1::uuid AND index_type='provider'`, event.AggregateID); err != nil {
+			return err
+		}
+		return p.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "system", Action: "memory.provider_delete.complete", ResourceType: "memory", ResourceID: &memoryPublicID, Result: "success", Detail: map[string]any{"provider_refs_deleted": len(refs)}})
+	})
+}
+
+type snapshotQueryer interface {
+	QueryRow(context.Context, string, ...any) pgx.Row
+}
+
+func (p *Processor) loadSnapshot(ctx context.Context, q snapshotQueryer, versionID string, lock bool) (providerSnapshot, error) {
+	lockClause := ""
+	if lock {
+		lockClause = " FOR UPDATE OF mi,cm,sp"
+	}
+	var s providerSnapshot
+	err := q.QueryRow(ctx, `SELECT t.id::text,t.public_id::text,h.public_id::text,sp.public_id::text,cm.public_id::text,mv.public_id::text,
+		mv.content,mv.content_digest,COALESCE(mi.provider_ref,''),cm.status,COALESCE(cm.current_version_id::text,''),cm.sensitivity,mv.version_no,
+		EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=cm.tenant_id AND cr.subject_id=cm.subject_id AND cr.consent_type='personalized_memory' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now()))
+		FROM memory_index_entry mi JOIN memory_version mv ON mv.id=mi.memory_version_id JOIN curated_memory cm ON cm.id=mi.memory_id
+		JOIN subject_profile sp ON sp.id=cm.subject_id AND sp.tenant_id=cm.tenant_id AND sp.status IN ('active','suspended')
+		JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id JOIN tenant t ON t.id=cm.tenant_id
+		WHERE mi.memory_version_id=$1::uuid AND mi.index_type='provider'`+lockClause, versionID).Scan(
+		&s.tenantInternalID, &s.tenantID, &s.householdID, &s.subjectID, &s.memoryID, &s.memoryVersionID, &s.content, &s.contentDigest, &s.providerRef, &s.status, &s.currentVersionID, &s.sensitivity, &s.versionNo, &s.consentEffective)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return providerSnapshot{}, errors.New("provider memory index entry not found")
+	}
+	return s, err
+}
+
+func providerEligible(s providerSnapshot, internalVersionID string) bool {
+	return s.status == "ACTIVE" && s.currentVersionID == internalVersionID && s.consentEffective && s.sensitivity == "normal"
+}
+
+func (p *Processor) deleteSupersededRefs(ctx context.Context, memoryPublicID, currentVersionID string) error {
+	rows, err := p.pool.Query(ctx, `SELECT memory_version_id::text,provider_ref FROM memory_index_entry mi JOIN curated_memory cm ON cm.id=mi.memory_id WHERE cm.public_id=$1::uuid AND mi.index_type='provider' AND mi.memory_version_id<>$2::uuid AND mi.provider_ref IS NOT NULL AND mi.status<>'DELETED'`, memoryPublicID, currentVersionID)
+	if err != nil {
+		return err
+	}
+	type ref struct{ versionID, providerRef string }
+	items := []ref{}
+	for rows.Next() {
+		var item ref
+		if err := rows.Scan(&item.versionID, &item.providerRef); err != nil {
+			rows.Close()
+			return err
+		}
+		items = append(items, item)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	for _, item := range items {
+		if err := p.client.Delete(ctx, item.providerRef, "provider-memory-supersede:"+item.versionID); err != nil {
+			return err
+		}
+	}
+	if len(items) > 0 {
+		_, err = p.pool.Exec(ctx, `UPDATE memory_index_entry mi SET status='DELETED',provider_ref=NULL,last_error=NULL,updated_at=now() FROM curated_memory cm WHERE mi.memory_id=cm.id AND cm.public_id=$1::uuid AND mi.index_type='provider' AND mi.memory_version_id<>$2::uuid`, memoryPublicID, currentVersionID)
+	}
+	return err
+}
+
+func (p *Processor) markFailed(ctx context.Context, versionID string, cause error) error {
+	message := cause.Error()
+	if len(message) > 450 {
+		message = message[:450]
+	}
+	_, _ = p.pool.Exec(ctx, `UPDATE memory_index_entry SET status='FAILED',last_attempt_at=now(),last_error=$2,updated_at=now() WHERE memory_version_id=$1::uuid AND index_type='provider'`, versionID, message)
+	if outbox.IsRetryable(cause) {
+		return cause
+	}
+	return fmt.Errorf("provider memory synchronization failed: %w", cause)
+}
@@ -0,0 +1,86 @@
+package providermemory
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"data-terminal-cloud/internal/outbox"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type recordingClient struct {
+	upserts []Memory
+	deletes []string
+}
+
+func (c *recordingClient) Upsert(_ context.Context, memory Memory) (string, error) {
+	c.upserts = append(c.upserts, memory)
+	return "provider-ref-" + memory.MemoryVersion, nil
+}
+func (c *recordingClient) Delete(_ context.Context, providerRef, _ string) error {
+	c.deletes = append(c.deletes, providerRef)
+	return nil
+}
+
+func TestProcessorSynchronizesAndDeletesProviderMemory(t *testing.T) {
+	databaseURL := os.Getenv("TEST_DATABASE_URL")
+	if databaseURL == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pool.Close()
+	memoryID, memoryPublicID, versionID, versionPublicID := uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()
+	defer func() {
+		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE resource_id=$1::uuid`, memoryPublicID)
+		_, _ = pool.Exec(ctx, `DELETE FROM memory_index_entry WHERE memory_id=$1::uuid`, memoryID)
+		_, _ = pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=NULL WHERE id=$1::uuid`, memoryID)
+		_, _ = pool.Exec(ctx, `DELETE FROM memory_version WHERE memory_id=$1::uuid`, memoryID)
+		_, _ = pool.Exec(ctx, `DELETE FROM curated_memory WHERE id=$1::uuid`, memoryID)
+	}()
+	if _, err := pool.Exec(ctx, `INSERT INTO curated_memory(id,public_id,tenant_id,subject_id,status,sensitivity,valid_from) VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401','ACTIVE','normal',now())`, memoryID, memoryPublicID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `INSERT INTO memory_version(id,public_id,memory_id,version_no,content,source_refs,confirmed_by,change_reason,content_digest) VALUES($1::uuid,$2::uuid,$3::uuid,1,'喜欢在周末照料兰花','[{"type":"integration"}]'::jsonb,'10000000-0000-0000-0000-000000000101','integration',repeat('a',64))`, versionID, versionPublicID, memoryID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=$2::uuid WHERE id=$1::uuid`, memoryID, versionID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status) VALUES('10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,$2::uuid,'provider','PENDING')`, memoryID, versionID); err != nil {
+		t.Fatal(err)
+	}
+
+	client := &recordingClient{}
+	processor, err := NewProcessor(pool, client)
+	if err != nil {
+		t.Fatal(err)
+	}
+	payload := []byte(`{"memory_version_id":"` + versionID + `"}`)
+	if err := processor.HandleSync(ctx, outbox.Event{AggregateID: memoryID, Payload: payload}); err != nil {
+		t.Fatal(err)
+	}
+	var status string
+	var providerRef *string
+	if err := pool.QueryRow(ctx, `SELECT status,provider_ref FROM memory_index_entry WHERE memory_version_id=$1::uuid AND index_type='provider'`, versionID).Scan(&status, &providerRef); err != nil {
+		t.Fatal(err)
+	}
+	if status != "ACTIVE" || providerRef == nil || len(client.upserts) != 1 || client.upserts[0].SubjectID != "10000000-0000-0000-0000-000000000411" {
+		t.Fatalf("status=%s provider=%v upserts=%+v", status, providerRef, client.upserts)
+	}
+	if err := processor.HandleDelete(ctx, outbox.Event{AggregateID: memoryID}); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT status,provider_ref FROM memory_index_entry WHERE memory_version_id=$1::uuid AND index_type='provider'`, versionID).Scan(&status, &providerRef); err != nil {
+		t.Fatal(err)
+	}
+	if status != "DELETED" || providerRef != nil || len(client.deletes) != 1 {
+		t.Fatalf("status=%s provider=%v deletes=%v", status, providerRef, client.deletes)
+	}
+}