package extraction

import (
	"context"
	"data-terminal-cloud/internal/outbox"
	"errors"
	"github.com/jackc/pgx/v5"
)

const ImportStoryEvent = "memoir.import_requested"

func loadImport(ctx context.Context, q queryer, id string) (extractionSnapshot, string, error) {
	var s extractionSnapshot
	var status, date string
	e := q.QueryRow(ctx, `SELECT ci.tenant_id::text,ci.subject_id::text,ci.status,to_char(ci.created_at AT TIME ZONE 'Asia/Shanghai','YYYY-MM-DD'),
 sp.status='active' AND NOT EXISTS(SELECT 1 FROM privacy_request pr WHERE pr.subject_id=sp.id AND pr.request_type='DELETE')
 AND NOT EXISTS(SELECT 1 FROM (VALUES('transcript_storage'),('story_extraction')) k(kind) WHERE NOT EXISTS(
 SELECT 1 FROM consent_record cr WHERE cr.subject_id=sp.id AND cr.consent_type=k.kind AND cr.withdrawn_at IS NULL
 AND cr.granted_at<=ci.created_at AND (cr.expires_at IS NULL OR cr.expires_at>now())))
 FROM memoir_chat_import ci JOIN subject_profile sp ON sp.id=ci.subject_id WHERE ci.id=$1::uuid`, id).Scan(&s.tenantID, &s.subjectID, &status, &date, &s.storyExtractionAllowed)
	if errors.Is(e, pgx.ErrNoRows) {
		return s, "", nil
	}
	if e != nil {
		return s, "", e
	}
	s.exists = true
	s.status = status
	rows, e := q.Query(ctx, `SELECT id::text,speaker,content FROM memoir_import_turn WHERE import_id=$1::uuid ORDER BY sequence_no`, id)
	if e != nil {
		return s, "", e
	}
	defer rows.Close()
	for rows.Next() {
		var t turn
		if e = rows.Scan(&t.id, &t.speaker, &t.text); e != nil {
			return s, "", e
		}
		t.publicID = t.id
		s.turns = append(s.turns, t)
	}
	s.digest = snapshotDigest(s)
	return s, date, rows.Err()
}
func (p *Processor) HandleImport(ctx context.Context, event outbox.Event) error {
	s, date, e := loadImport(ctx, p.pool, event.AggregateID)
	if e != nil {
		return e
	}
	if !s.exists || s.status == "READY" {
		return nil
	}
	if !s.storyExtractionAllowed {
		return errors.New("import consent unavailable")
	}
	result, previous, e := p.generateEventsSafely(ctx, s, date)
	if e != nil {
		return e
	}
	tx, e := p.pool.Begin(ctx)
	if e != nil {
		return e
	}
	defer tx.Rollback(context.Background())
	if e = lockStoryContext(ctx, tx, s, previous); e != nil {
		return e
	}
	if _, e = tx.Exec(ctx, `SELECT id FROM memoir_chat_import WHERE id=$1::uuid FOR UPDATE`, event.AggregateID); e != nil {
		return e
	}
	current, _, e := loadImport(ctx, tx, event.AggregateID)
	if e != nil {
		return e
	}
	if current.status == "READY" {
		return tx.Commit(ctx)
	}
	if !current.storyExtractionAllowed || current.digest != s.digest {
		return outbox.Retryable(ErrConversationContentChanged)
	}
	if e = storeEventStories(ctx, tx, "", "", event.AggregateID, 0, current, result, 1, previous); e != nil {
		return e
	}
	_, e = tx.Exec(ctx, `UPDATE memoir_chat_import SET status='READY' WHERE id=$1::uuid`, event.AggregateID)
	if e != nil {
		return e
	}
	return tx.Commit(ctx)
}