package callmemory

import (
	"context"
	"data-terminal-cloud/internal/identity"
	"encoding/base64"
	"errors"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/google/uuid"
)

func queryMemoryIDs(t *testing.T, result map[string]any) []string {
	t.Helper()
	ids := []string{}
	for _, raw := range result["items"].([]any) {
		item := raw.(map[string]any)
		ids = append(ids, item["memory_id"].(string))
		if _, err := time.Parse(time.RFC3339Nano, item["created_at"].(string)); err != nil {
			t.Fatalf("invalid created_at: %v", item)
		}
	}
	return ids
}

func TestMemoryQueryCreatedAtPagination(t *testing.T) {
	f := setup(t)
	base := time.Date(2026, 10, 2, 10, 0, 0, 123456000, time.UTC)
	ids := make([]string, 5)
	for i := range ids {
		ids[i], _ = f.add(t)
		// The first three share a microsecond timestamp. The older records have
		// newer updated_at values, which must not move them to the first page.
		created := base
		if i >= 3 {
			created = base.Add(-time.Duration(i) * time.Hour)
		}
		f.exec(t, `UPDATE curated_memory SET created_at=$2,updated_at=$3 WHERE public_id=$1::uuid`, ids[i], created, base.Add(time.Duration(i)*time.Hour))
	}
	want := append([]string{}, ids[:3]...)
	sort.Sort(sort.Reverse(sort.StringSlice(want)))
	want = append(want, ids[3:]...)
	in := QueryInput{ElderID: f.elder, Limit: 2}
	first := f.request(t, "query", in)
	got := queryMemoryIDs(t, first)
	if !reflect.DeepEqual(got, want[:2]) || !strings.HasPrefix(first["next_cursor"].(string), "v1.") {
		t.Fatalf("first page=%v want=%v cursor=%v", got, want[:2], first["next_cursor"])
	}
	// The legacy UUID anchor resolves within the authorized list and gives the
	// same chronological continuation as the new timestamp/ID cursor.
	in.Cursor = got[1]
	legacy := f.request(t, "query", in)
	in.Cursor = first["next_cursor"].(string)
	second := f.request(t, "query", in)
	if !reflect.DeepEqual(queryMemoryIDs(t, legacy), queryMemoryIDs(t, second)) {
		t.Fatalf("legacy=%v new=%v", legacy, second)
	}
	// A new arrival and an edit cannot shift the saved page boundary.
	newID, _ := f.add(t)
	f.exec(t, `UPDATE curated_memory SET created_at=$2 WHERE public_id=$1::uuid`, newID, base.Add(time.Hour))
	f.exec(t, `UPDATE curated_memory SET updated_at=$2 WHERE public_id=$1::uuid`, ids[4], base.Add(24*time.Hour))
	// A self-contained cursor also survives removal of its anchor from the list.
	f.exec(t, `UPDATE curated_memory SET status='DELETED' WHERE public_id=$1::uuid`, got[1])
	second = f.request(t, "query", in)
	got = append(got, queryMemoryIDs(t, second)...)
	in.Cursor = second["next_cursor"].(string)
	if in.Cursor == "" {
		t.Fatal("missing second-page cursor")
	}
	last := f.request(t, "query", in)
	got = append(got, queryMemoryIDs(t, last)...)
	if !reflect.DeepEqual(got, want) || last["next_cursor"] != "" {
		t.Fatalf("pages=%v want=%v last=%v", got, want, last)
	}
}

func TestMemoryQueryCursorScopeAndValidation(t *testing.T) {
	f, other := setup(t), setup(t)
	id, _ := f.add(t)
	foreignID, _ := other.add(t)
	base := time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)
	f.exec(t, `UPDATE curated_memory SET created_at=$2 WHERE public_id=$1::uuid`, id, base)
	bad := []QueryInput{
		{ElderID: f.elder, Cursor: foreignID},
		{ElderID: f.elder, Cursor: uuid.NewString()},
		{ElderID: f.elder, Cursor: id, Type: "RELATIONSHIP"},
		{ElderID: f.elder, Cursor: id, Status: "DISABLED"},
		{ElderID: f.elder, Cursor: "invalid"},
		{ElderID: f.elder, Cursor: "v1.invalid"},
		{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString([]byte(`{"created_at":"invalid","id":"`+id+`"}`))},
		{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString(jsonBytes(memoryQueryCursor{CreatedAt: base, ID: "invalid"}))},
		{ElderID: f.elder, Cursor: strings.Repeat("x", 1025)},
	}
	for _, in := range bad {
		_, err := f.s.MemoryRequest(context.Background(), f.principal, "query", jsonBytes(in))
		var typed *Error
		if !errors.As(err, &typed) || typed.Code != "INVALID_CURSOR" {
			t.Fatalf("cursor=%q type=%q status=%q error=%v", in.Cursor, in.Type, in.Status, err)
		}
	}
	// Even a valid timestamp/ID boundary cannot return another elder's records.
	in := QueryInput{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString(jsonBytes(memoryQueryCursor{CreatedAt: base.Add(time.Hour), ID: foreignID}))}
	if got := queryMemoryIDs(t, f.request(t, "query", in)); !reflect.DeepEqual(got, []string{id}) {
		t.Fatalf("cross-elder cursor returned %v", got)
	}
	_, err := f.s.MemoryRequest(context.Background(), identity.Principal{Issuer: f.principal.Issuer, Subject: uuid.NewString()}, "query", jsonBytes(QueryInput{ElderID: f.elder, Cursor: id}))
	var typed *Error
	if !errors.As(err, &typed) || typed.Code != "MEMORY_ACCESS_DENIED" {
		t.Fatalf("unrelated account query error=%v", err)
	}
	// A delegated reader cannot use a hidden sensitive record as a legacy anchor.
	reader := identity.Principal{Issuer: f.principal.Issuer, Subject: uuid.NewString()}
	user := uuid.NewString()
	f.exec(t, `INSERT INTO user_account(id,tenant_id,oidc_subject,oidc_issuer) VALUES($1::uuid,$2::uuid,$3,$4)`, user, f.tenant, reader.Subject, reader.Issuer)
	f.exec(t, `INSERT INTO household_member(tenant_id,household_id,user_id,role,status) VALUES($1::uuid,'10000000-0000-0000-0000-000000000201',$2::uuid,'viewer','active')`, f.tenant, user)
	f.exec(t, `INSERT INTO memory_access_grant(subject_id,user_id,can_read,can_write,sensitive_read) VALUES($1::uuid,$2::uuid,true,false,false)`, f.subject, user)
	f.exec(t, `UPDATE curated_memory SET sensitivity='sensitive' WHERE public_id=$1::uuid`, id)
	_, err = f.s.MemoryRequest(context.Background(), reader, "query", jsonBytes(QueryInput{ElderID: f.elder, Cursor: id}))
	if !errors.As(err, &typed) || typed.Code != "INVALID_CURSOR" {
		t.Fatalf("hidden sensitive cursor error=%v", err)
	}
}

func addStatusCall(t *testing.T, f fixture, ended bool, epoch int64, turns ...[2]string) string {
	t.Helper()
	sid := uuid.NewString()
	f.exec(t, `INSERT INTO call_session(id,tenant_id,subject_id,device_id,boot_id,hook_event_id,call_generation,binding_generation,call_state,transcript_state,transcript_revision,memory_mode,privacy_epoch,ended_at,budget_snapshot) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$1::uuid,$1,1,1,CASE WHEN $5::boolean THEN 'ENDED' ELSE 'TALKING' END,'PARTIAL',$7,'HOT_ONLY',$6,CASE WHEN $5::boolean THEN now() END,'{"protocol":"DIRECT-CHAT-1"}')`, sid, f.tenant, f.subject, f.device, ended, epoch, len(turns))
	for _, turn := range turns {
		f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,$2,0,$3,$4,false,now())`, sid, uuid.NewString(), turn[0], turn[1])
	}
	return sid
}

func TestMemorySyncStatusInsufficientTranscript(t *testing.T) {
	f, other := setup(t), setup(t)
	f.s.ConfigureManagedSync(func(context.Context) error { return nil })
	memoryID, _ := f.add(t)
	f.exec(t, `UPDATE memory_user_state SET privacy_epoch=3 WHERE subject_id=$1::uuid`, f.subject)
	user, assistant := [2]string{"USER", "今天走了两公里"}, [2]string{"ASSISTANT", "散步后记得休息"}
	empty := addStatusCall(t, f, true, 3)
	f.exec(t, `UPDATE call_session SET transcript_state='EMPTY' WHERE id=$1::uuid`, empty)
	missingUser := addStatusCall(t, f, true, 3, assistant, [2]string{"USER", " \t\n "})
	addStatusCall(t, f, true, 3, user)
	addStatusCall(t, f, true, 3, [2]string{"UNKNOWN", "无法归属的文字"})
	addStatusCall(t, f, true, 3, assistant, [2]string{"USER", "[deleted]"})
	// Missing calls are counted once whether no job exists or several transcript
	// revisions were skipped; SKIPPED itself is not the classification signal.
	f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,0,'SKIPPED'),($2::uuid,1,'SKIPPED'),($2::uuid,2,'SKIPPED')`, empty, missingUser)
	completeSkipped := addStatusCall(t, f, true, 3, user, assistant)
	f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,2,'SKIPPED')`, completeSkipped)
	addStatusCall(t, f, false, 3, assistant)
	draining := addStatusCall(t, f, true, 3, assistant)
	f.exec(t, `UPDATE call_session SET transcript_state='DRAINING',drain_started_at=now() WHERE id=$1::uuid`, draining)
	collecting := addStatusCall(t, f, true, 3)
	f.exec(t, `UPDATE call_session SET transcript_state='COLLECTING' WHERE id=$1::uuid`, collecting)
	addStatusCall(t, f, true, 2, assistant) // an old privacy epoch
	addStatusCall(t, other, true, 0, assistant)
	legacy := addStatusCall(t, f, true, 3, assistant)
	f.exec(t, `UPDATE call_session SET budget_snapshot='{"protocol":"CALL-MEMORY-1.1"}' WHERE id=$1::uuid`, legacy)
	off := addStatusCall(t, f, true, 3, assistant)
	f.exec(t, `UPDATE call_session SET memory_mode='OFF' WHERE id=$1::uuid`, off)
	suppressed := addStatusCall(t, f, true, 3, assistant)
	f.exec(t, `INSERT INTO memory_source_suppression(subject_id,memory_id,source_refs,privacy_epoch) SELECT $1::uuid,id,jsonb_build_array(jsonb_build_object('session_id',$3::text)),3 FROM curated_memory WHERE public_id=$2::uuid`, f.subject, memoryID, suppressed)
	for _, state := range []string{"PENDING", "UPLOADED", "SYNCED"} {
		sid := addStatusCall(t, f, true, 3, user, assistant)
		f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,1,$2),($1::uuid,2,$2)`, sid, state)
	}
	status := f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
	for key, want := range map[string]float64{"insufficient_transcript_count": 5, "upload_pending_count": 1, "extracting_count": 1, "synced_call_count": 1} {
		if status[key] != want {
			t.Fatalf("%s=%v want=%v status=%v", key, status[key], want, status)
		}
	}
	// Late valid text fixes the classification without needing a worker or retry.
	f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,'late-user',0,'USER','我也去散步了',false,now())`, missingUser)
	status = f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
	if status["insufficient_transcript_count"] != float64(4) {
		t.Fatalf("late text status=%v", status)
	}
}

func TestMemorySyncStatusInsufficientRespectsAuthorization(t *testing.T) {
	for _, consent := range []string{"personalized_memory", "transcript_storage"} {
		for _, expire := range []bool{false, true} {
			name := consent + "/withdrawn"
			if expire {
				name = consent + "/expired"
			}
			t.Run(name, func(t *testing.T) {
				f := setup(t)
				addStatusCall(t, f, true, 0)
				in := map[string]any{"elder_id": f.elder}
				if status := f.request(t, "sync/status", in); status["insufficient_transcript_count"] != float64(1) {
					t.Fatalf("authorized status=%v", status)
				}
				if expire {
					f.exec(t, `UPDATE consent_record SET granted_at=now()-interval '1 day',expires_at=now()-interval '1 second' WHERE subject_id=$1::uuid AND consent_type=$2`, f.subject, consent)
				} else {
					f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type=$2`, f.subject, consent)
				}
				// Match the new epoch so authorization, independently of epoch filtering,
				// must suppress these calls from the missing-transcript count.
				f.exec(t, `UPDATE call_session SET privacy_epoch=(SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid) WHERE subject_id=$1::uuid`, f.subject)
				if status := f.request(t, "sync/status", in); status["insufficient_transcript_count"] != float64(0) {
					t.Fatalf("unauthorized status=%v", status)
				}
			})
		}
	}
}