story/internal/providermemory/processor.go

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)
}