package elderagent

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

const personaVoiceAllowedSQL = `p.status='active' AND h.status='active'
 AND COALESCE((SELECT effective_consents @> '{"voice_chat":true}'::jsonb FROM consent_snapshot WHERE subject_id=p.id ORDER BY snapshot_version DESC LIMIT 1),false)
 AND NOT EXISTS(SELECT 1 FROM privacy_request WHERE subject_id=p.id AND request_type='DELETE' AND status NOT IN ('FAILED','REJECTED','CANCELLED'))`

const personaActiveSQL = `EXISTS(SELECT 1 FROM call_session c WHERE c.subject_id=p.id
 AND (c.manifest_status IN ('ISSUED','REVOKED') OR c.budget_snapshot->>'protocol'='DIRECT-CHAT-1')
 AND NOT(c.local_audio_off AND c.sdk_stopped AND (c.remote_state IN ('NOT_STARTED','STOPPED_OBSERVED')
 OR (COALESCE(c.budget_snapshot->>'protocol','')='DIRECT-CHAT-1' AND c.ended_at IS NOT NULL))))
 OR EXISTS(SELECT 1 FROM conversation c WHERE c.subject_id=p.id AND c.status IN ('ROOM_ISSUED','ACTIVE','STOPPING'))`

type personaSnapshot struct {
	ID, PublicID, Bot, Base, Name, Language, State, Digest, AgentStatus string
	Config, Agent, Profile, Preferences                                 []byte
	Revision, Synced                                                    int64
	Attempts                                                            int
	Voice, Memory, PublishedVoice, PublishedMemory, Active, Dirty       bool
}

type personaQuerier interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

func loadPersona(ctx context.Context, q personaQuerier, subject string) (personaSnapshot, error) {
	var p personaSnapshot
	err := q.QueryRow(ctx, `SELECT p.id::text,p.public_id::text,COALESCE(a.bot_id,''),a.persona_base_role,p.preferred_name,p.language,
 a.config,a.agent_config,p.profile_data,p.conversation_preferences,a.persona_revision,a.persona_synced_revision,
 a.persona_sync_state,a.persona_sync_attempts,a.persona_digest,(`+personaVoiceAllowedSQL+`),call_memory_authorized(p.id),
 a.persona_voice_allowed,a.persona_memory_allowed,(`+personaActiveSQL+`),a.persona_session_dirty,a.status
 FROM subject_voice_agent a JOIN subject_profile p ON p.id=a.subject_id JOIN household h ON h.id=p.household_id
 WHERE p.id=$1::uuid`, subject).Scan(&p.ID, &p.PublicID, &p.Bot, &p.Base, &p.Name, &p.Language, &p.Config, &p.Agent, &p.Profile, &p.Preferences, &p.Revision, &p.Synced, &p.State, &p.Attempts, &p.Digest, &p.Voice, &p.Memory, &p.PublishedVoice, &p.PublishedMemory, &p.Active, &p.Dirty, &p.AgentStatus)
	p.Memory = p.Memory && p.Voice
	return p, err
}

func (p personaSnapshot) configuration() (map[string]any, map[string]any, error) {
	role, err := compilePersona(p.Base, p.Name, p.Language, p.Profile, p.Preferences, p.Voice, p.Memory)
	if err != nil {
		return nil, nil, err
	}
	return personaConfiguration(p.Config, p.Agent, role)
}

func configurationDigest(config, agent map[string]any) string {
	raw, _ := json.Marshal([]any{config, agent})
	sum := sha256.Sum256(raw)
	return hex.EncodeToString(sum[:])
}

func (s *Service) verify(ctx context.Context, bot string, config, agent map[string]any) error {
	// The production client reads the provider configuration back; embedded test
	// providers can use the smaller provisioning interface.
	if verifier, ok := s.Provider.(interface {
		VerifyConfiguration(context.Context, string, map[string]any, map[string]any) error
	}); ok {
		return verifier.VerifyConfiguration(ctx, bot, config, agent)
	}
	return nil
}

func unlockPersona(conn *pgxpool.Conn, id string) {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if _, err := conn.Exec(ctx, `SELECT pg_advisory_unlock(hashtextextended($1,64))`, id); err != nil {
		_ = conn.Conn().Close(ctx)
	}
}

// Sync attempts immediately without delaying a save behind an active preparation.
// The revision and retry time persist independently of this request's lifetime.
func (s *Service) Sync(ctx context.Context, publicID string) (Info, error) {
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	conn, err := s.Pool.Acquire(ctx)
	if err != nil {
		return Info{}, err
	}
	defer conn.Release()
	var id string
	if err = conn.QueryRow(ctx, `SELECT id::text FROM subject_profile WHERE public_id=$1::uuid`, publicID).Scan(&id); err != nil {
		return Info{}, err
	}
	var locked bool
	if err = conn.QueryRow(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,64))`, id).Scan(&locked); err != nil {
		return Info{}, err
	}
	if !locked {
		return s.Get(ctx, publicID)
	}
	defer unlockPersona(conn, id)
	err = s.syncLocked(ctx, conn, id)
	info, getErr := s.Get(ctx, publicID)
	if getErr != nil {
		return info, getErr
	}
	return info, err
}

func (s *Service) syncLocked(ctx context.Context, conn *pgxpool.Conn, id string) error {
	p, err := loadPersona(ctx, conn, id)
	if err != nil {
		return err
	}
	if p.AgentStatus != "READY" {
		return nil
	}
	if p.Active {
		_, err = conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_sync_state=CASE WHEN persona_revision<>persona_synced_revision OR persona_voice_allowed<>$2 OR persona_memory_allowed<>$3 THEN 'DEFERRED' ELSE persona_sync_state END,persona_next_retry_at=now()+interval '5 seconds' WHERE subject_id=$1::uuid`, id, p.Voice, p.Memory)
		return err
	}
	config, agent, err := p.configuration()
	if err == nil {
		err = s.configureNativeMemory(ctx, id, config)
	}
	if err != nil {
		return s.recordPersonaFailure(conn, p, err)
	}
	digest := configurationDigest(config, agent)
	// Equality includes all provider settings, not only the prompt.
	if p.Revision == p.Synced && digest == p.Digest && p.Voice == p.PublishedVoice && p.Memory == p.PublishedMemory && !p.Dirty {
		return nil
	}
	updater, ok := s.Provider.(interface {
		Update(context.Context, string, map[string]any, map[string]any) error
	})
	if !ok {
		return s.recordPersonaFailure(conn, p, fmt.Errorf("AIBOT_UPDATE_REQUIRED"))
	}
	if err = updater.Update(ctx, p.Bot, config, agent); err == nil {
		err = s.verifyNativeMemory(ctx, p.Bot, config)
	}
	if err == nil {
		err = s.verify(ctx, p.Bot, config, agent)
	}
	if err != nil {
		return s.recordPersonaFailure(conn, p, err)
	}
	// A concurrent consent change may advance the revision while the external call
	// runs. Never acknowledge that newer version, and leave it queued for cleanup.
	_, err = conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_synced_revision=$2,persona_sync_state='SYNCED',persona_sync_attempts=0,persona_error_code='',persona_synced_at=now(),persona_digest=$3,persona_memory_allowed=$4,persona_voice_allowed=$5,persona_session_dirty=false WHERE subject_id=$1::uuid AND persona_revision=$2`, id, p.Revision, digest, p.Memory, p.Voice)
	return err
}

func (s *Service) recordPersonaFailure(conn *pgxpool.Conn, p personaSnapshot, cause error) error {
	// Provider errors may contain credentials or user text. Persist a fixed code.
	attempt := min(p.Attempts, 8)
	delay := min(time.Second*time.Duration(1<<attempt), 5*time.Minute)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	_, err := conn.Exec(ctx, `UPDATE subject_voice_agent SET persona_sync_state='FAILED',persona_sync_attempts=LEAST(persona_sync_attempts+1,1000000),persona_error_code='PERSONA_SYNC_FAILED',persona_next_retry_at=$3 WHERE subject_id=$1::uuid AND persona_revision=$2`, p.ID, p.Revision, time.Now().Add(delay))
	if err != nil {
		return err
	}
	return fmt.Errorf("PERSONA_SYNC_FAILED")
}

// RunPersonaSync also cleans call-specific context after a call has fully stopped,
// and detects permission expiry without needing a new user edit.
func (s *Service) RunPersonaSync(ctx context.Context, logger *slog.Logger) {
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		if err := s.SyncPending(ctx); err != nil && ctx.Err() == nil {
			logger.Warn("elder persona sync scan failed")
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (s *Service) SyncPending(ctx context.Context) error {
	rows, err := s.Pool.Query(ctx, `SELECT p.public_id::text FROM subject_voice_agent a JOIN subject_profile p ON p.id=a.subject_id JOIN household h ON h.id=p.household_id
 WHERE a.status='READY' AND a.persona_next_retry_at<=now() AND (a.persona_sync_state<>'SYNCED' OR a.persona_session_dirty
 OR a.persona_memory_allowed IS DISTINCT FROM (call_memory_authorized(p.id) AND (`+personaVoiceAllowedSQL+`)) OR a.persona_voice_allowed IS DISTINCT FROM (`+personaVoiceAllowedSQL+`)) ORDER BY a.persona_next_retry_at,a.subject_id LIMIT 10`)
	if err != nil {
		return err
	}
	ids := []string{}
	for rows.Next() {
		var id string
		if err = rows.Scan(&id); err != nil {
			rows.Close()
			return err
		}
		ids = append(ids, id)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	for _, id := range ids {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		_, _ = s.Sync(ctx, id)
	}
	return nil
}