package callmemory

import (
	"context"
	"errors"
	"fmt"
	"github.com/jackc/pgx/v5"
	"time"
)

type ProjectionRecord struct {
	Tenant, Subject, ID, Kind, Content, Policy, Status, ItemStatus string
	Version                                                        int64
}
type ProjectionResult struct {
	Kind, ID string
	Visible  bool
}

// ReconcileOnly forbids another create after an uncertain acknowledgement.
type ProjectionProvider interface {
	Project(context.Context, ProjectionRecord, string, string, bool) (ProjectionResult, error)
}
type SourceTurn struct {
	ID       string    `json:"utterance_id"`
	Revision int64     `json:"source_revision"`
	Text     string    `json:"text"`
	Time     time.Time `json:"source_at"`
}
type Intent struct {
	Utterance  string      `json:"utterance_id"`
	Revision   int64       `json:"source_revision"`
	Action     string      `json:"action"`
	Targets    []string    `json:"targets"`
	Patch      MemoryPatch `json:"patch"`
	Confidence float64     `json:"confidence"`
	Quote      string      `json:"source_quote"`
	Start      int         `json:"source_start"`
	End        int         `json:"source_end"`
}
type IntentExtractor interface {
	ExtractMemoryIntents(context.Context, string, []SourceTurn, []map[string]any) ([]Intent, error)
}

func (s *Service) ConfigureProjection(p ProjectionProvider)              { s.projection = p }
func (s *Service) ConfigureExtractor(e IntentExtractor)                  { s.extractor = e }
func (s *Service) ConfigureManagedSync(sync func(context.Context) error) { s.managedSync = sync }
func (s *Service) Run(ctx context.Context) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		if e := s.Maintain(ctx); e != nil && ctx.Err() == nil {
			s.logger.Error("call memory maintenance failed")
		}
		if s.managedSync != nil {
			if e := s.managedSync(ctx); e != nil && ctx.Err() == nil {
				s.logger.Warn("managed memory synchronization pending", "code", e.Error())
			}
		}
		if e := s.ProjectNext(ctx); e != nil && ctx.Err() == nil {
			s.logger.Warn("memory projection pending")
		}
		if e := s.ExtractNext(ctx); e != nil && ctx.Err() == nil {
			s.logger.Warn("memory extraction pending")
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}
func (s *Service) Maintain(ctx context.Context) error {
	tx, e := s.pool.Begin(ctx)
	if e != nil {
		return e
	}
	defer tx.Rollback(context.Background())
	// Direct chat has a bounded local lease. Missing end is an interrupted call,
	// never evidence that the supplier finished or that every subtitle arrived.
	_, e = tx.Exec(ctx, `UPDATE call_session SET call_state='FAILED',manifest_status='DISCARDED',manifest_b64=NULL,ended_at=now(),drain_started_at=now(),transcript_state='DRAINING',text_gap=true WHERE budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND ended_at IS NULL AND (lease_expires_at<=now() OR expires_at<=now())`)
	if e != nil {
		return e
	}
	// Clock-based expiry must revoke projections even without a consent UPDATE.
	_, e = tx.Exec(ctx, `UPDATE curated_memory SET status='DISABLED',version=version+1,updated_at=now() WHERE status='ACTIVE' AND NOT call_memory_authorized(subject_id)`)
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `WITH revoked AS (
 UPDATE call_session SET manifest_status='REVOKED',call_state=CASE WHEN call_state IN ('PREPARING','STARTING','TALKING') THEN 'STOPPING' ELSE call_state END
 WHERE manifest_status='ISSUED' AND lease_expires_at<=now() RETURNING id,privacy_epoch)
 INSERT INTO call_command(session_id,privacy_epoch,type,reason,expires_at) SELECT id,privacy_epoch,'STOP_CALL','LEASE_EXPIRED',now()+interval '7 days' FROM revoked ON CONFLICT DO NOTHING`)
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE call_session c SET transcript_state=CASE
 WHEN text_gap OR EXISTS(SELECT 1 FROM call_text_fragment f WHERE f.session_id=c.id GROUP BY utterance_id,source_revision HAVING count(*)<max(part_count)) THEN 'PARTIAL'
 WHEN EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id) THEN 'COMPLETE' ELSE 'EMPTY' END
 WHERE transcript_state='DRAINING' AND (drain_started_at<=now()-interval '60 seconds' OR GREATEST(COALESCE(last_text_at,drain_started_at),drain_started_at)<=now()-interval '15 seconds')`)
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `INSERT INTO call_postprocess_job(session_id,transcript_revision,extractor_version) SELECT id,transcript_revision,'memory-1' FROM call_session WHERE transcript_state IN ('COMPLETE','PARTIAL') AND transcript_revision>0 AND (NOT $1::boolean OR COALESCE(budget_snapshot->>'protocol','')<>'DIRECT-CHAT-1') ON CONFLICT DO NOTHING`, s.managedSync != nil)
	if e != nil {
		return e
	}
	if s.managedSync != nil {
		if _, e = tx.Exec(ctx, `UPDATE call_postprocess_job j SET state='SUPERSEDED',last_error=NULL FROM call_session c WHERE c.id=j.session_id AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND j.state='PENDING'`); e != nil {
			return e
		}
	}
	if e = s.enqueueDirectStories(ctx, tx); e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state='EXPIRED',confirmation_hash=NULL,revision=revision+1,updated_at=now() WHERE state='NEEDS_CONFIRMATION' AND confirmation_expires_at<=now()`)
	if e != nil {
		return e
	}
	if e = s.advanceMutations(ctx, tx); e != nil {
		return e
	}
	// Receipts can expire; deletion ledgers and suppressions must survive cleanup.
	_, e = tx.Exec(ctx, `DELETE FROM call_request_result WHERE expires_at<now()`)
	if e != nil {
		return e
	}
	return tx.Commit(ctx)
}
func (s *Service) advanceMutations(ctx context.Context, tx pgx.Tx) error {
	rows, e := tx.Query(ctx, `SELECT id::text,subject_id::text,action,scope,COALESCE(memory_revision,0),COALESCE(privacy_epoch,0),targets FROM memory_mutation WHERE state IN ('PROPAGATING','BLOCKED') ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 50`)
	if e != nil {
		return e
	}
	type work struct {
		id, subject, action, scope string
		rev, epoch                 int64
		targets                    []byte
	}
	jobs := []work{}
	for rows.Next() {
		var w work
		if e = rows.Scan(&w.id, &w.subject, &w.action, &w.scope, &w.rev, &w.epoch, &w.targets); e != nil {
			rows.Close()
			return e
		}
		jobs = append(jobs, w)
	}
	e = rows.Err()
	rows.Close()
	if e != nil {
		return e
	}
	for _, w := range jobs {
		var visible int64
		var legacy, active, projectionBlocked bool
		e = tx.QueryRow(ctx, `SELECT viking_visible_revision,legacy_projection_pending,EXISTS(SELECT 1 FROM call_session WHERE subject_id=$1::uuid AND privacy_epoch<$2 AND manifest_status IN ('ISSUED','REVOKED') AND NOT(local_audio_off AND sdk_stopped AND remote_state IN ('NOT_STARTED','STOPPED_OBSERVED'))) OR call_legacy_cleanup_pending($1::uuid),EXISTS(SELECT 1 FROM memory_projection_job WHERE subject_id=$1::uuid AND memory_revision<=$3 AND state='BLOCKED') FROM memory_user_state WHERE subject_id=$1::uuid`, w.subject, w.epoch, w.rev).Scan(&visible, &legacy, &active, &projectionBlocked)
		if e != nil {
			return e
		}
		stage := map[string]string{"main_store": "DONE", "serving_projection": "PENDING", "derived_context": "DONE", "active_context": "PENDING", "source_cleanup": "DONE", "backup_cleanup": "DONE"}
		if visible >= w.rev && !legacy {
			stage["serving_projection"] = "DONE"
		}
		if !active {
			stage["active_context"] = "DONE"
		}
		if w.action == "ADD" {
			stage["derived_context"], stage["active_context"] = "NOT_REQUIRED", "NOT_REQUIRED"
		}
		if w.action != "DELETE" {
			stage["source_cleanup"], stage["backup_cleanup"] = "NOT_REQUIRED", "NOT_REQUIRED"
		}
		if w.action == "DELETE" {
			if e = s.cleanSources(ctx, tx, w.subject, w.targets, w.scope); e != nil {
				return e
			}
			var pendingBackup bool
			e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM memory_deletion_ledger l JOIN curated_memory cm ON cm.id=l.memory_id WHERE cm.subject_id=$1::uuid AND cm.public_id IN (SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)) AND l.backup_cleanup_verified_at IS NULL)`, w.subject, w.targets).Scan(&pendingBackup)
			if e != nil {
				return e
			}
			if pendingBackup {
				stage["backup_cleanup"] = "PENDING"
			}
		}
		state := "PROPAGATING"
		complete := true
		for _, v := range stage {
			if v != "DONE" && v != "NOT_REQUIRED" {
				complete = false
			}
		}
		if complete {
			state = "COMPLETED"
			if w.action == "DELETE" {
				if _, e = tx.Exec(ctx, `UPDATE curated_memory SET status='DELETED',updated_at=now() WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)) AND status='PENDING_DELETE'`, w.subject, w.targets); e != nil {
					return e
				}
			}
		}
		if projectionBlocked {
			state = "BLOCKED"
			stage["serving_projection"] = "BLOCKED"
		}
		_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state=$2,stage_states=$3::jsonb,revision=revision+CASE WHEN state<>$2 OR stage_states<>$3::jsonb THEN 1 ELSE 0 END,updated_at=now() WHERE id=$1::uuid`, w.id, state, jsonBytes(stage))
		if e != nil {
			return e
		}
	}
	return nil
}
func (s *Service) cleanSources(ctx context.Context, tx pgx.Tx, subject string, targets []byte, scope string) error {
	// Preserve only tombstones. Source-suppression carries IDs and provenance, not plaintext.
	_, e := tx.Exec(ctx, `UPDATE memory_version SET content='[deleted]',change_reason=NULL,content_digest=$3 WHERE memory_id IN (SELECT id FROM curated_memory WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb))) AND content<>'[deleted]'`, subject, targets, hash([]byte("[deleted]")))
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE curated_memory SET entity_key='',completion_basis=CASE WHEN completion_basis IS NOT NULL THEN '{"redacted":true}'::jsonb END WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb))`, subject, targets)
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='DELETED',updated_at=now() WHERE memory_id IN(SELECT id FROM curated_memory WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)))`, subject, targets)
	if e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE memory_mutation m SET patch='{}'::jsonb,confirmation_hash=NULL,proposal_digest=NULL,confirmation_expires_at=NULL,state=CASE WHEN state IN ('RECEIVED','NEEDS_CONFIRMATION') THEN 'CANCELLED' ELSE state END WHERE subject_id=$1::uuid AND EXISTS(SELECT 1 FROM jsonb_array_elements_text(m.targets) a JOIN jsonb_array_elements_text($2::jsonb) b ON a.value=b.value)`, subject, targets)
	if e != nil {
		return e
	}
	if scope == "SOURCE_AND_DERIVED" {
		if e = s.cleanStorySources(ctx, tx, subject, targets); e != nil {
			return e
		}
	}
	// Old derived summaries cannot continue carrying a removed fact.
	_, e = tx.Exec(ctx, `UPDATE conversation_summary SET content='[deleted]',status='SUPERSEDED' WHERE subject_id=$1::uuid AND status='SUPERSEDED' AND content<>'[deleted]'`, subject)
	return e
}
func (s *Service) ProjectNext(ctx context.Context) error {
	if s.projection == nil {
		return nil
	}
	conn, e := s.pool.Acquire(ctx)
	if e != nil {
		return e
	}
	defer conn.Release()
	var subject string
	e = conn.QueryRow(ctx, `SELECT j.subject_id::text FROM memory_projection_job j WHERE j.state NOT IN ('VISIBLE','BLOCKED') AND j.next_attempt_at<=now() AND j.memory_revision=(SELECT min(x.memory_revision) FROM memory_projection_job x WHERE x.subject_id=j.subject_id AND x.state<>'VISIBLE') ORDER BY j.next_attempt_at,j.created_at LIMIT 1`).Scan(&subject)
	if errors.Is(e, pgx.ErrNoRows) {
		return nil
	}
	if e != nil {
		return e
	}
	var locked bool
	e = conn.QueryRow(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,13))`, subject).Scan(&locked)
	if e != nil || !locked {
		return e
	}
	defer conn.Exec(context.Background(), `SELECT pg_advisory_unlock(hashtextextended($1,13))`, subject)
	var jobID, state, priorID, priorKind string
	var r ProjectionRecord
	var jobVersion, revision int64
	e = conn.QueryRow(ctx, `SELECT j.id::text,j.state,j.record_version,j.memory_revision,cm.tenant_id::text,cm.subject_id::text,cm.public_id::text,cm.memory_type,mv.content,cm.proactive_policy,CASE WHEN cm.sensitivity='normal' AND call_memory_authorized(cm.subject_id) THEN cm.status ELSE 'DISABLED' END,COALESCE(cm.item_status,''),cm.version,COALESCE(mp.provider_id,''),COALESCE(mp.provider_kind,'') FROM memory_projection_job j JOIN curated_memory cm ON cm.id=j.memory_id JOIN memory_version mv ON mv.id=cm.current_version_id LEFT JOIN memory_projection mp ON mp.memory_id=cm.id WHERE j.subject_id=$1::uuid AND j.state<>'VISIBLE' ORDER BY j.memory_revision LIMIT 1`, subject).Scan(&jobID, &state, &jobVersion, &revision, &r.Tenant, &r.Subject, &r.ID, &r.Kind, &r.Content, &r.Policy, &r.Status, &r.ItemStatus, &r.Version, &priorID, &priorKind)
	if e != nil {
		return e
	}
	if state == "BLOCKED" {
		return nil
	}
	// Project current authoritative state even if this job predates a correction.
	// A critical job is never skipped without verifying the resulting projection.
	reconcile := state == "WRITING" || state == "RECONCILING" || state == "WRITE_ACKED"
	if _, e = conn.Exec(ctx, `UPDATE memory_projection_job SET state='WRITING',attempts=attempts+1 WHERE id=$1::uuid`, jobID); e != nil {
		return e
	}
	result, writeErr := s.projection.Project(ctx, r, priorKind, priorID, reconcile)
	nextState := "WRITE_ACKED"
	code := ""
	if writeErr != nil {
		nextState = "RECONCILING"
		code = "PROVIDER_RESULT_UNKNOWN"
		var f *Error
		if errors.As(writeErr, &f) && f.Status < 500 {
			nextState = "BLOCKED"
			code = f.Code
		}
	} else if result.Visible {
		nextState = "VISIBLE"
	}
	tx, e := conn.Begin(ctx)
	if e != nil {
		return e
	}
	defer tx.Rollback(context.Background())
	if _, e = tx.Exec(ctx, `SELECT subject_id FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, subject); e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE memory_projection_job SET state=$2,provider_kind=NULLIF($3,''),provider_id=NULLIF($4,''),last_error=NULLIF($5,''),next_attempt_at=now()+interval '5 seconds' WHERE id=$1::uuid`, jobID, nextState, result.Kind, result.ID, code)
	if e != nil {
		return e
	}
	if writeErr == nil && result.ID != "" {
		_, e = tx.Exec(ctx, `INSERT INTO memory_projection(memory_id,subject_id,provider_kind,provider_id,record_version,visible_version,operation_key) SELECT id,subject_id,$2,$3,$4,$5,$6 FROM curated_memory WHERE public_id=$1::uuid ON CONFLICT(memory_id) DO UPDATE SET provider_kind=excluded.provider_kind,provider_id=excluded.provider_id,record_version=excluded.record_version,visible_version=excluded.visible_version,operation_key=excluded.operation_key,updated_at=now()`, r.ID, result.Kind, result.ID, r.Version, map[bool]int64{true: r.Version, false: 0}[result.Visible], fmt.Sprintf("%s:%d", r.ID, r.Version))
		if e != nil {
			return e
		}
	}
	// A concurrent correction can commit while the external write is in flight.
	// Reconcile that version before acknowledging visibility of this job.
	var currentVersion int64
	var currentStatus string
	if e = tx.QueryRow(ctx, `SELECT version,CASE WHEN sensitivity='normal' AND call_memory_authorized(subject_id) THEN status ELSE 'DISABLED' END FROM curated_memory WHERE public_id=$1::uuid FOR UPDATE`, r.ID).Scan(&currentVersion, &currentStatus); e != nil {
		return e
	}
	if currentVersion != r.Version || currentStatus != r.Status {
		nextState = "RECONCILING"
		if _, e = tx.Exec(ctx, `UPDATE memory_projection_job SET state='RECONCILING',next_attempt_at=now() WHERE id=$1::uuid`, jobID); e != nil {
			return e
		}
	}
	if nextState == "VISIBLE" {
		_, e = tx.Exec(ctx, `UPDATE memory_user_state us SET viking_visible_revision=COALESCE((SELECT min(memory_revision)-1 FROM memory_projection_job j WHERE j.subject_id=us.subject_id AND j.state<>'VISIBLE'),us.memory_revision),updated_at=now() WHERE subject_id=$1::uuid`, subject)
		if e != nil {
			return e
		}
	}
	if e = tx.Commit(ctx); e != nil {
		return e
	}
	return writeErr
}