package controlplane

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"database/sql"

	_ "modernc.org/sqlite"
)

var (
	ErrAccountBindingConflict  = errors.New("account binding conflicts with the registered source")
	ErrAccountNotAuthorized    = errors.New("account is not currently authorized for data ingestion")
	ErrAccountStoreNotFound    = errors.New("account data store was not found")
	ErrBatchConflict           = errors.New("ingest batch id or payload conflicts with an existing batch")
	ErrBatchSequenceGap        = errors.New("ingest batch sequence has a gap")
	ErrBatchSequenceConflict   = errors.New("ingest batch sequence conflicts with confirmed progress")
	ErrAccountCapacityExceeded = errors.New("account data capacity budget exceeded")
)

const (
	accountSchemaVersion = "1"
	catalogSchemaVersion = "1"
)

// AccountStoreManager owns the small catalog and opens one platform-owned
// SQLite shard per verified account. The existing JSON Store remains the
// source of truth for control-plane tasks, nodes, events and audit records.
type AccountStoreManagerOptions struct {
	Retention     time.Duration
	MaxShardBytes int64
	MaxBatchBytes int64
}

type AccountStoreManager struct {
	root          string
	catalog       *sql.DB
	retention     time.Duration
	maxShardBytes int64
	maxBatchBytes int64
	mu            sync.Mutex
	closed        bool
}

type AccountRegistration struct {
	AccountID              string
	StableIdentity         string
	SourceNodeID           string
	SourceGeneration       string
	Verified               bool
	AuthorizationVersion   int
	AuthorizationExpiresAt *time.Time
	ReportingScopes        []ReportingScope
}

type ReportingScope struct {
	ChatID        string
	DataType      string
	ExpiresAt     *time.Time
	ConfigVersion int
}

type ConversationRecord struct {
	ChatID         string
	ChatType       string
	Title          string
	LastActivityAt *time.Time
	Source         string
	ObservedAt     time.Time
	DirectoryState string
}

type MessageRecord struct {
	MessageID       string
	ChatID          string
	SourceMessageID string
	Direction       string
	MessageType     string
	Text            string
	SourceTime      time.Time
	ObservedAt      time.Time
	SourceVersion   string
	PayloadHash     string
}

type IngestBatch struct {
	BatchID          string
	SourceGeneration string
	StreamKey        string
	Sequence         int64
	CursorStart      string
	CursorEnd        string
	PayloadHash      string
	CoverageState    string
	Conversations    []ConversationRecord
	Messages         []MessageRecord
}

type BatchApplyResult struct {
	Duplicate         bool
	ConfirmedSequence int64
	ConfirmedCursor   string
}

type AccountStore struct {
	manager       *AccountStoreManager
	accountID     string
	path          string
	db            *sql.DB
	maxShardBytes int64
	maxBatchBytes int64
	closeOnce     sync.Once
	closeErr      error
}

func OpenAccountStoreManager(root string, options ...AccountStoreManagerOptions) (*AccountStoreManager, error) {
	if strings.TrimSpace(root) == "" {
		return nil, errors.New("account data directory is required")
	}
	cleanRoot := filepath.Clean(root)
	if err := os.MkdirAll(cleanRoot, 0o700); err != nil {
		return nil, fmt.Errorf("create account data directory: %w", err)
	}
	catalog, err := openSQLite(filepath.Join(cleanRoot, "catalog.sqlite"))
	if err != nil {
		return nil, err
	}
	var configured AccountStoreManagerOptions
	if len(options) > 0 {
		configured = options[0]
	}
	if configured.Retention < 0 || configured.MaxShardBytes < 0 || configured.MaxBatchBytes < 0 {
		_ = catalog.Close()
		return nil, errors.New("account store budgets must be non-negative")
	}
	manager := &AccountStoreManager{
		root: cleanRoot, catalog: catalog,
		retention: configured.Retention, maxShardBytes: configured.MaxShardBytes, maxBatchBytes: configured.MaxBatchBytes,
	}
	if err := manager.initCatalog(context.Background()); err != nil {
		_ = catalog.Close()
		return nil, err
	}
	return manager, nil
}

func (m *AccountStoreManager) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.closed {
		return nil
	}
	m.closed = true
	if m.catalog == nil {
		return nil
	}
	return m.catalog.Close()
}

func (m *AccountStoreManager) RegisterAccount(ctx context.Context, registration AccountRegistration) (*AccountStore, error) {
	if err := validateRegistration(registration); err != nil {
		return nil, err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.closed {
		return nil, errors.New("account store manager is closed")
	}

	var existing struct {
		sourceNodeID, sourceGeneration, stableIdentity, shardPath string
		verified, authorizationVersion                            int
		authorizationExpiresAt                                    sql.NullString
	}
	err := m.catalog.QueryRowContext(ctx, `
		SELECT source_node_id, source_generation, stable_identity, shard_path,
		       verified, authorization_version, authorization_expires_at
		FROM accounts WHERE account_id = ?`, registration.AccountID).
		Scan(&existing.sourceNodeID, &existing.sourceGeneration, &existing.stableIdentity,
			&existing.shardPath, &existing.verified, &existing.authorizationVersion,
			&existing.authorizationExpiresAt)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf("read account registration: %w", err)
	}

	shardPath := existing.shardPath
	created := errors.Is(err, sql.ErrNoRows)
	if created {
		shardPath = accountShardPath(m.root, registration.AccountID)
		if err := createAccountShard(shardPath); err != nil {
			return nil, err
		}
	} else if existing.sourceGeneration != registration.SourceGeneration || existing.stableIdentity != registration.StableIdentity {
		return nil, ErrAccountBindingConflict
	}

	tx, err := m.catalog.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin account registration: %w", err)
	}
	rollback := func(cause error) (*AccountStore, error) {
		_ = tx.Rollback()
		return nil, cause
	}
	expiresAt := nullableTimeArg(registration.AuthorizationExpiresAt)
	now := formatTime(ptrTime(time.Now().UTC()))
	if created {
		_, err = tx.ExecContext(ctx, `
			INSERT INTO accounts (
				account_id, source_node_id, source_generation, stable_identity,
				verified, authorization_version, authorization_expires_at,
				shard_path, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			registration.AccountID, registration.SourceNodeID, registration.SourceGeneration,
			registration.StableIdentity, boolInt(registration.Verified), registration.AuthorizationVersion,
			expiresAt, shardPath, now, now)
	} else {
		_, err = tx.ExecContext(ctx, `
			UPDATE accounts
		SET verified = ?, authorization_version = ?, authorization_expires_at = ?, updated_at = ?
		WHERE account_id = ?`,
			boolInt(registration.Verified), registration.AuthorizationVersion, expiresAt, now, registration.AccountID)
	}
	if err != nil {
		return rollback(fmt.Errorf("write account registration: %w", err))
	}
	if registration.ReportingScopes != nil {
		if _, err := tx.ExecContext(ctx, `DELETE FROM reporting_scopes WHERE account_id = ?`, registration.AccountID); err != nil {
			return rollback(fmt.Errorf("replace reporting scopes: %w", err))
		}
		for _, scope := range registration.ReportingScopes {
			if err := validateScope(scope); err != nil {
				return rollback(err)
			}
			if _, err := tx.ExecContext(ctx, `
				INSERT INTO reporting_scopes (account_id, chat_id, data_type, expires_at, config_version)
				VALUES (?, ?, ?, ?, ?)`, registration.AccountID, scope.ChatID, scope.DataType,
				nullableTimeArg(scope.ExpiresAt), scope.ConfigVersion); err != nil {
				return rollback(fmt.Errorf("write reporting scope: %w", err))
			}
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit account registration: %w", err)
	}
	return m.openAccountLocked(ctx, registration.AccountID, shardPath)
}

func (m *AccountStoreManager) OpenAccount(ctx context.Context, accountID string) (*AccountStore, error) {
	if strings.TrimSpace(accountID) == "" {
		return nil, errors.New("account id is required")
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.closed {
		return nil, errors.New("account store manager is closed")
	}
	var shardPath string
	if err := m.catalog.QueryRowContext(ctx, `SELECT shard_path FROM accounts WHERE account_id = ?`, accountID).Scan(&shardPath); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, ErrAccountStoreNotFound
		}
		return nil, fmt.Errorf("find account shard: %w", err)
	}
	return m.openAccountLocked(ctx, accountID, shardPath)
}

func (m *AccountStoreManager) openAccountLocked(ctx context.Context, accountID, shardPath string) (*AccountStore, error) {
	db, err := openSQLite(shardPath)
	if err != nil {
		return nil, err
	}
	if err := validateAccountSchema(ctx, db); err != nil {
		_ = db.Close()
		return nil, err
	}
	if err := ensureAccountIndexes(ctx, db); err != nil {
		_ = db.Close()
		return nil, err
	}
	return &AccountStore{
		manager: m, accountID: accountID, path: shardPath, db: db,
		maxShardBytes: m.maxShardBytes, maxBatchBytes: m.maxBatchBytes,
	}, nil
}

func (m *AccountStoreManager) initCatalog(ctx context.Context) error {
	_, err := m.catalog.ExecContext(ctx, `
		CREATE TABLE IF NOT EXISTS schema_meta (
			key TEXT PRIMARY KEY,
			value TEXT NOT NULL
		);
		CREATE TABLE IF NOT EXISTS accounts (
			account_id TEXT PRIMARY KEY,
			source_node_id TEXT NOT NULL,
			source_generation TEXT NOT NULL,
			stable_identity TEXT NOT NULL,
			verified INTEGER NOT NULL CHECK (verified IN (0, 1)),
			authorization_version INTEGER NOT NULL,
			authorization_expires_at TEXT,
			shard_path TEXT NOT NULL UNIQUE,
			created_at TEXT NOT NULL,
			updated_at TEXT NOT NULL
		);
		CREATE TABLE IF NOT EXISTS reporting_scopes (
			account_id TEXT NOT NULL REFERENCES accounts(account_id) ON DELETE CASCADE,
			chat_id TEXT NOT NULL,
			data_type TEXT NOT NULL,
			expires_at TEXT,
			config_version INTEGER NOT NULL,
			PRIMARY KEY (account_id, chat_id, data_type)
		);
		INSERT INTO schema_meta(key, value) VALUES ('schema_version', '1')
		ON CONFLICT(key) DO UPDATE SET value = excluded.value;
	`)
	if err != nil {
		return fmt.Errorf("initialize account catalog: %w", err)
	}
	return nil
}

func (s *AccountStore) Close() error {
	s.closeOnce.Do(func() { s.closeErr = s.db.Close() })
	return s.closeErr
}

func (s *AccountStore) Path() string      { return s.path }
func (s *AccountStore) AccountID() string { return s.accountID }

type MaintenanceReport struct {
	Accounts             int
	DeletedMessages      int64
	DeletedConversations int64
	DeletedBatches       int64
	Checkpointed         int
	OverBudgetAccounts   []string
	Errors               []string
}

// RunMaintenance applies the configured retention policy and performs a
// passive WAL checkpoint for each account. One broken shard is reported and
// isolated so it cannot prevent maintenance of other accounts.
func (m *AccountStoreManager) RunMaintenance(ctx context.Context, now time.Time) (MaintenanceReport, error) {
	if now.IsZero() {
		now = time.Now().UTC()
	} else {
		now = now.UTC()
	}
	m.mu.Lock()
	if m.closed {
		m.mu.Unlock()
		return MaintenanceReport{}, errors.New("account store manager is closed")
	}
	rows, err := m.catalog.QueryContext(ctx, `SELECT account_id FROM accounts ORDER BY account_id`)
	if err != nil {
		m.mu.Unlock()
		return MaintenanceReport{}, fmt.Errorf("list account shards: %w", err)
	}
	var accountIDs []string
	for rows.Next() {
		var accountID string
		if err := rows.Scan(&accountID); err != nil {
			_ = rows.Close()
			m.mu.Unlock()
			return MaintenanceReport{}, fmt.Errorf("scan account shard: %w", err)
		}
		accountIDs = append(accountIDs, accountID)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		m.mu.Unlock()
		return MaintenanceReport{}, fmt.Errorf("read account shards: %w", err)
	}
	_ = rows.Close()
	m.mu.Unlock()

	report := MaintenanceReport{Accounts: len(accountIDs)}
	for _, accountID := range accountIDs {
		store, err := m.OpenAccount(ctx, accountID)
		if err != nil {
			report.Errors = append(report.Errors, accountID+":open")
			continue
		}
		deletedMessages, deletedConversations, deletedBatches, err := store.maintain(ctx, now, m.retention)
		if err != nil {
			report.Errors = append(report.Errors, accountID+":maintenance")
			_ = store.Close()
			continue
		}
		report.DeletedMessages += deletedMessages
		report.DeletedConversations += deletedConversations
		report.DeletedBatches += deletedBatches
		report.Checkpointed++
		if m.maxShardBytes > 0 {
			if size, sizeErr := accountStorageBytes(store.path); sizeErr != nil {
				report.Errors = append(report.Errors, accountID+":size")
			} else if size > m.maxShardBytes {
				report.OverBudgetAccounts = append(report.OverBudgetAccounts, accountID)
			}
		}
		_ = store.Close()
	}
	return report, nil
}

func (s *AccountStore) maintain(ctx context.Context, now time.Time, retention time.Duration) (int64, int64, int64, error) {
	var deletedMessages, deletedConversations, deletedBatches int64
	if retention > 0 {
		cutoff := formatTime(ptrTime(now.Add(-retention)))
		result, err := s.db.ExecContext(ctx, `DELETE FROM messages WHERE observed_at < ?`, cutoff)
		if err != nil {
			return 0, 0, 0, fmt.Errorf("retain messages: %w", err)
		}
		deletedMessages, _ = result.RowsAffected()
		result, err = s.db.ExecContext(ctx, `DELETE FROM conversations WHERE observed_at < ? AND NOT EXISTS (SELECT 1 FROM messages WHERE messages.chat_id = conversations.chat_id)`, cutoff)
		if err != nil {
			return 0, 0, 0, fmt.Errorf("retain conversations: %w", err)
		}
		deletedConversations, _ = result.RowsAffected()
		result, err = s.db.ExecContext(ctx, `DELETE FROM ingest_batches WHERE status = 'confirmed' AND confirmed_at IS NOT NULL AND confirmed_at < ?`, cutoff)
		if err != nil {
			return 0, 0, 0, fmt.Errorf("retain ingest batches: %w", err)
		}
		deletedBatches, _ = result.RowsAffected()
	}
	if _, err := s.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE)`); err != nil {
		return 0, 0, 0, fmt.Errorf("checkpoint account database: %w", err)
	}
	if _, err := s.db.ExecContext(ctx, `PRAGMA optimize`); err != nil {
		return 0, 0, 0, fmt.Errorf("optimize account database: %w", err)
	}
	return deletedMessages, deletedConversations, deletedBatches, nil
}

// RestoreAccount validates a consistent account backup before atomically
// replacing the registered shard. The catalog binding is retained, so a
// restored file cannot change the account/source identity.
func (m *AccountStoreManager) RestoreAccount(ctx context.Context, accountID, backupPath string) error {
	if strings.TrimSpace(accountID) == "" || strings.TrimSpace(backupPath) == "" {
		return errors.New("account id and backup path are required")
	}
	m.mu.Lock()
	if m.closed {
		m.mu.Unlock()
		return errors.New("account store manager is closed")
	}
	var shardPath string
	if err := m.catalog.QueryRowContext(ctx, `SELECT shard_path FROM accounts WHERE account_id = ?`, accountID).Scan(&shardPath); err != nil {
		m.mu.Unlock()
		if errors.Is(err, sql.ErrNoRows) {
			return ErrAccountStoreNotFound
		}
		return fmt.Errorf("find account shard: %w", err)
	}
	m.mu.Unlock()

	source, err := os.Open(filepath.Clean(backupPath))
	if err != nil {
		return fmt.Errorf("open account backup: %w", err)
	}
	defer source.Close()
	temporary, err := os.CreateTemp(filepath.Dir(shardPath), ".restore-*")
	if err != nil {
		return fmt.Errorf("create restore file: %w", err)
	}
	temporaryPath := temporary.Name()
	defer os.Remove(temporaryPath)
	if _, err := io.Copy(temporary, source); err != nil {
		_ = temporary.Close()
		return fmt.Errorf("copy account backup: %w", err)
	}
	if err := temporary.Sync(); err != nil {
		_ = temporary.Close()
		return fmt.Errorf("sync restored account: %w", err)
	}
	if err := temporary.Close(); err != nil {
		return fmt.Errorf("close restored account: %w", err)
	}
	if err := os.Chmod(temporaryPath, 0o600); err != nil {
		return fmt.Errorf("protect restored account: %w", err)
	}
	check, err := openSQLite(temporaryPath)
	if err != nil {
		return fmt.Errorf("open restored account: %w", err)
	}
	if err := validateAccountSchema(ctx, check); err != nil {
		_ = check.Close()
		return err
	}
	if err := ensureAccountIndexes(ctx, check); err != nil {
		_ = check.Close()
		return err
	}
	var integrity string
	if err := check.QueryRowContext(ctx, `PRAGMA integrity_check`).Scan(&integrity); err != nil {
		_ = check.Close()
		return fmt.Errorf("check restored account: %w", err)
	}
	if integrity != "ok" {
		_ = check.Close()
		return fmt.Errorf("restored account integrity check returned %q", integrity)
	}
	if err := check.Close(); err != nil {
		return fmt.Errorf("close restored account: %w", err)
	}
	oldPath := shardPath + fmt.Sprintf(".before-restore-%d", time.Now().UnixNano())
	if err := os.Rename(shardPath, oldPath); err != nil {
		return fmt.Errorf("stage current account shard: %w", err)
	}
	if err := os.Rename(temporaryPath, shardPath); err != nil {
		_ = os.Rename(oldPath, shardPath)
		return fmt.Errorf("install restored account shard: %w", err)
	}
	if err := os.Remove(oldPath); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("remove old account shard: %w", err)
	}
	return nil
}

// Backup creates a consistent SQLite backup without copying a live main file
// while WAL frames are pending. The destination must not already exist.
func (s *AccountStore) Backup(ctx context.Context, targetPath string) error {
	if strings.TrimSpace(targetPath) == "" {
		return errors.New("backup path is required")
	}
	cleanPath := filepath.Clean(targetPath)
	if _, err := os.Stat(cleanPath); err == nil {
		return errors.New("backup path already exists")
	} else if !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("check backup path: %w", err)
	}
	if err := os.MkdirAll(filepath.Dir(cleanPath), 0o700); err != nil {
		return fmt.Errorf("create backup directory: %w", err)
	}
	if _, err := s.db.ExecContext(ctx, "VACUUM INTO ?", cleanPath); err != nil {
		return fmt.Errorf("backup account database: %w", err)
	}
	if err := os.Chmod(cleanPath, 0o600); err != nil {
		return fmt.Errorf("protect account backup: %w", err)
	}
	return nil
}

func (s *AccountStore) ApplyBatch(ctx context.Context, batch IngestBatch) (BatchApplyResult, error) {
	if err := validateBatch(batch); err != nil {
		return BatchApplyResult{}, err
	}
	scopes, sourceGeneration, err := s.manager.activeScopes(ctx, s.accountID)
	if err != nil {
		return BatchApplyResult{}, err
	}
	if sourceGeneration != batch.SourceGeneration {
		return BatchApplyResult{}, ErrAccountBindingConflict
	}
	if err := authorizeBatch(batch, scopes); err != nil {
		return BatchApplyResult{}, err
	}
	if err := s.ensureCapacity(batch); err != nil {
		return BatchApplyResult{}, err
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return BatchApplyResult{}, fmt.Errorf("begin ingest batch: %w", err)
	}
	rollback := func(cause error) (BatchApplyResult, error) {
		_ = tx.Rollback()
		return BatchApplyResult{}, cause
	}
	var existingHash, existingGeneration, existingStatus string
	var existingSequence int64
	err = tx.QueryRowContext(ctx, `SELECT payload_hash, source_generation, sequence, status FROM ingest_batches WHERE batch_id = ?`, batch.BatchID).
		Scan(&existingHash, &existingGeneration, &existingSequence, &existingStatus)
	if err == nil {
		if existingHash != batch.PayloadHash || existingGeneration != batch.SourceGeneration || existingSequence != batch.Sequence {
			return rollback(ErrBatchConflict)
		}
		if existingStatus == "confirmed" {
			if err := tx.Rollback(); err != nil {
				return BatchApplyResult{}, fmt.Errorf("close duplicate ingest transaction: %w", err)
			}
			return BatchApplyResult{Duplicate: true, ConfirmedSequence: existingSequence, ConfirmedCursor: batch.CursorEnd}, nil
		}
	} else if !errors.Is(err, sql.ErrNoRows) {
		return rollback(fmt.Errorf("read ingest batch: %w", err))
	}

	var confirmedSequence int64
	var currentGeneration string
	err = tx.QueryRowContext(ctx, `SELECT source_generation, confirmed_sequence FROM sync_state WHERE stream_key = ?`, batch.StreamKey).
		Scan(&currentGeneration, &confirmedSequence)
	if errors.Is(err, sql.ErrNoRows) {
		currentGeneration = batch.SourceGeneration
		confirmedSequence = 0
	} else if err != nil {
		return rollback(fmt.Errorf("read sync state: %w", err))
	} else if currentGeneration != batch.SourceGeneration {
		return rollback(ErrAccountBindingConflict)
	}
	if batch.Sequence > confirmedSequence+1 {
		return rollback(ErrBatchSequenceGap)
	}
	if batch.Sequence <= confirmedSequence && existingStatus != "confirmed" {
		return rollback(ErrBatchSequenceConflict)
	}

	now := formatTime(ptrTime(time.Now().UTC()))
	if existingStatus == "" {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO ingest_batches (batch_id, source_generation, stream_key, sequence, cursor_start, cursor_end, payload_hash, status, received_at)
			VALUES (?, ?, ?, ?, ?, ?, ?, 'applying', ?)`, batch.BatchID, batch.SourceGeneration, batch.StreamKey,
			batch.Sequence, batch.CursorStart, batch.CursorEnd, batch.PayloadHash, now); err != nil {
			return rollback(fmt.Errorf("record ingest batch: %w", err))
		}
	}
	for _, conversation := range batch.Conversations {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO conversations (chat_id, chat_type, title, last_activity_at, source, observed_at, directory_state)
			VALUES (?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(chat_id) DO UPDATE SET
				chat_type = excluded.chat_type,
				title = excluded.title,
				last_activity_at = excluded.last_activity_at,
				source = excluded.source,
				observed_at = excluded.observed_at,
				directory_state = excluded.directory_state`, conversation.ChatID, conversation.ChatType,
			conversation.Title, formatTime(conversation.LastActivityAt), conversation.Source,
			formatTime(ptrTime(conversation.ObservedAt)), conversation.DirectoryState); err != nil {
			return rollback(fmt.Errorf("write conversation: %w", err))
		}
	}
	for _, message := range batch.Messages {
		if _, err := tx.ExecContext(ctx, `
			INSERT INTO messages (
				message_id, chat_id, source_message_id, direction, message_type, text,
				source_time, observed_at, source_version, payload_hash
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(message_id) DO UPDATE SET
				chat_id = excluded.chat_id,
				source_message_id = excluded.source_message_id,
				direction = excluded.direction,
				message_type = excluded.message_type,
				text = excluded.text,
				source_time = excluded.source_time,
				observed_at = excluded.observed_at,
				source_version = excluded.source_version,
				payload_hash = excluded.payload_hash`, message.MessageID, message.ChatID,
			message.SourceMessageID, message.Direction, message.MessageType, message.Text,
			formatTime(ptrTime(message.SourceTime)), formatTime(ptrTime(message.ObservedAt)),
			message.SourceVersion, message.PayloadHash); err != nil {
			return rollback(fmt.Errorf("write message: %w", err))
		}
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO sync_state (stream_key, source_generation, confirmed_sequence, confirmed_cursor, coverage_state, last_success_at)
		VALUES (?, ?, ?, ?, ?, ?)
		ON CONFLICT(stream_key) DO UPDATE SET
			source_generation = excluded.source_generation,
			confirmed_sequence = excluded.confirmed_sequence,
			confirmed_cursor = excluded.confirmed_cursor,
			coverage_state = excluded.coverage_state,
			last_success_at = excluded.last_success_at,
			error_code = NULL,
			error_message = NULL,
			updated_at = excluded.last_success_at`, batch.StreamKey, batch.SourceGeneration,
		batch.Sequence, batch.CursorEnd, defaultCoverage(batch.CoverageState), now); err != nil {
		return rollback(fmt.Errorf("write sync state: %w", err))
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE ingest_batches SET status = 'confirmed', confirmed_at = ? WHERE batch_id = ?`, now, batch.BatchID); err != nil {
		return rollback(fmt.Errorf("confirm ingest batch: %w", err))
	}
	if err := tx.Commit(); err != nil {
		return BatchApplyResult{}, fmt.Errorf("commit ingest batch: %w", err)
	}
	return BatchApplyResult{ConfirmedSequence: batch.Sequence, ConfirmedCursor: batch.CursorEnd}, nil
}

func (m *AccountStoreManager) activeScopes(ctx context.Context, accountID string) ([]ReportingScope, string, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.closed {
		return nil, "", errors.New("account store manager is closed")
	}
	var verified int
	var generation string
	var expires sql.NullString
	if err := m.catalog.QueryRowContext(ctx, `SELECT verified, source_generation, authorization_expires_at FROM accounts WHERE account_id = ?`, accountID).Scan(&verified, &generation, &expires); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, "", ErrAccountStoreNotFound
		}
		return nil, "", fmt.Errorf("read account authorization: %w", err)
	}
	if verified == 0 || (expires.Valid && strings.TrimSpace(expires.String) != "" && parseTime(expires.String).Before(time.Now().UTC())) {
		return nil, "", ErrAccountNotAuthorized
	}
	rows, err := m.catalog.QueryContext(ctx, `SELECT chat_id, data_type, expires_at, config_version FROM reporting_scopes WHERE account_id = ?`, accountID)
	if err != nil {
		return nil, "", fmt.Errorf("read reporting scopes: %w", err)
	}
	defer rows.Close()
	var scopes []ReportingScope
	for rows.Next() {
		var scope ReportingScope
		var expiresAt sql.NullString
		if err := rows.Scan(&scope.ChatID, &scope.DataType, &expiresAt, &scope.ConfigVersion); err != nil {
			return nil, "", fmt.Errorf("scan reporting scope: %w", err)
		}
		if expiresAt.Valid && strings.TrimSpace(expiresAt.String) != "" {
			expiry := parseTime(expiresAt.String)
			scope.ExpiresAt = &expiry
		}
		if scope.ExpiresAt == nil || scope.ExpiresAt.After(time.Now().UTC()) {
			scopes = append(scopes, scope)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, "", fmt.Errorf("read reporting scopes: %w", err)
	}
	return scopes, generation, nil
}

func (s *AccountStore) QueryConversations(ctx context.Context, limit, offset int) ([]ConversationRecord, error) {
	if limit < 1 || limit > 200 || offset < 0 {
		return nil, errors.New("invalid conversation pagination")
	}
	scopes, _, err := s.manager.activeScopes(ctx, s.accountID)
	if err != nil {
		return nil, err
	}
	if len(scopes) == 0 {
		return nil, ErrAccountNotAuthorized
	}
	rows, err := s.db.QueryContext(ctx, `SELECT chat_id, chat_type, title, last_activity_at, source, observed_at, directory_state FROM conversations ORDER BY COALESCE(last_activity_at, observed_at) DESC, chat_id`)
	if err != nil {
		return nil, fmt.Errorf("query conversations: %w", err)
	}
	defer rows.Close()
	var records []ConversationRecord
	for rows.Next() {
		var item ConversationRecord
		var lastActivity, observed string
		if err := rows.Scan(&item.ChatID, &item.ChatType, &item.Title, &lastActivity, &item.Source, &observed, &item.DirectoryState); err != nil {
			return nil, err
		}
		if lastActivity != "" {
			value := parseTime(lastActivity)
			item.LastActivityAt = &value
		}
		item.ObservedAt = parseTime(observed)
		if scopeAllows(scopes, item.ChatID, "conversations") {
			records = append(records, item)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if offset >= len(records) {
		return []ConversationRecord{}, nil
	}
	end := offset + limit
	if end > len(records) {
		end = len(records)
	}
	return records[offset:end], nil
}

type AccountSyncStatus struct {
	StreamKey         string
	SourceGeneration  string
	ConfirmedSequence int64
	ConfirmedCursor   string
	CoverageState     string
	LastSuccessAt     *time.Time
	ErrorCode         string
	ErrorMessage      string
}

func (s *AccountStore) AuthorizeChat(ctx context.Context, chatID, dataType string) (bool, error) {
	scopes, _, err := s.manager.activeScopes(ctx, s.accountID)
	if err != nil {
		return false, err
	}
	return scopeAllows(scopes, chatID, dataType), nil
}

func (s *AccountStore) ensureCapacity(batch IngestBatch) error {
	estimated := estimateBatchBytes(batch)
	if s.maxBatchBytes > 0 && estimated > s.maxBatchBytes {
		return fmt.Errorf("%w: batch exceeds byte budget", ErrAccountCapacityExceeded)
	}
	if s.maxShardBytes <= 0 {
		return nil
	}
	current, err := accountStorageBytes(s.path)
	if err != nil {
		return fmt.Errorf("check account capacity: %w", err)
	}
	if current+estimated > s.maxShardBytes {
		return fmt.Errorf("%w: shard exceeds byte budget", ErrAccountCapacityExceeded)
	}
	return nil
}

func (s *AccountStore) GetSyncStatus(ctx context.Context, streamKey string) (AccountSyncStatus, error) {
	if strings.TrimSpace(streamKey) == "" {
		return AccountSyncStatus{}, errors.New("sync stream key is required")
	}
	var status AccountSyncStatus
	var lastSuccess, errorCode, errorMessage sql.NullString
	err := s.db.QueryRowContext(ctx, `SELECT stream_key, source_generation, confirmed_sequence, confirmed_cursor, coverage_state, last_success_at, error_code, error_message FROM sync_state WHERE stream_key = ?`, streamKey).
		Scan(&status.StreamKey, &status.SourceGeneration, &status.ConfirmedSequence, &status.ConfirmedCursor, &status.CoverageState, &lastSuccess, &errorCode, &errorMessage)
	if errors.Is(err, sql.ErrNoRows) {
		return AccountSyncStatus{StreamKey: streamKey, CoverageState: "unknown"}, nil
	}
	if err != nil {
		return AccountSyncStatus{}, fmt.Errorf("query sync status: %w", err)
	}
	if lastSuccess.Valid && strings.TrimSpace(lastSuccess.String) != "" {
		value := parseTime(lastSuccess.String)
		status.LastSuccessAt = &value
	}
	if errorCode.Valid {
		status.ErrorCode = errorCode.String
	}
	if errorMessage.Valid {
		status.ErrorMessage = errorMessage.String
	}
	return status, nil
}

func (s *AccountStore) QueryMessages(ctx context.Context, chatID string, limit, offset int) ([]MessageRecord, error) {
	if strings.TrimSpace(chatID) == "" || limit < 1 || limit > 200 || offset < 0 {
		return nil, errors.New("invalid message query")
	}
	allowed, err := s.AuthorizeChat(ctx, chatID, "messages")
	if err != nil {
		return nil, err
	}
	if !allowed {
		return nil, ErrAccountNotAuthorized
	}
	rows, err := s.db.QueryContext(ctx, `SELECT message_id, chat_id, source_message_id, direction, message_type, text, source_time, observed_at, source_version, payload_hash FROM messages WHERE chat_id = ? ORDER BY source_time DESC, message_id DESC LIMIT ? OFFSET ?`, chatID, limit, offset)
	if err != nil {
		return nil, fmt.Errorf("query messages: %w", err)
	}
	defer rows.Close()
	var records []MessageRecord
	for rows.Next() {
		var item MessageRecord
		var sourceTime, observed string
		if err := rows.Scan(&item.MessageID, &item.ChatID, &item.SourceMessageID, &item.Direction, &item.MessageType, &item.Text, &sourceTime, &observed, &item.SourceVersion, &item.PayloadHash); err != nil {
			return nil, err
		}
		item.SourceTime = parseTime(sourceTime)
		item.ObservedAt = parseTime(observed)
		records = append(records, item)
	}
	return records, rows.Err()
}

func (m *AccountStoreManager) SetScopes(ctx context.Context, accountID string, scopes []ReportingScope) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.closed {
		return errors.New("account store manager is closed")
	}
	tx, err := m.catalog.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM reporting_scopes WHERE account_id = ?`, accountID); err != nil {
		_ = tx.Rollback()
		return err
	}
	for _, scope := range scopes {
		if err := validateScope(scope); err != nil {
			_ = tx.Rollback()
			return err
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO reporting_scopes (account_id, chat_id, data_type, expires_at, config_version) VALUES (?, ?, ?, ?, ?)`, accountID, scope.ChatID, scope.DataType, nullableTimeArg(scope.ExpiresAt), scope.ConfigVersion); err != nil {
			_ = tx.Rollback()
			return err
		}
	}
	return tx.Commit()
}

func validateAccountSchema(ctx context.Context, db *sql.DB) error {
	var version string
	if err := db.QueryRowContext(ctx, `SELECT value FROM schema_meta WHERE key = 'schema_version'`).Scan(&version); err != nil {
		return fmt.Errorf("read account schema version: %w", err)
	}
	if version != accountSchemaVersion {
		return fmt.Errorf("unsupported account schema version %q", version)
	}
	return nil
}

func ensureAccountIndexes(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx, `
		CREATE INDEX IF NOT EXISTS idx_messages_chat_time ON messages(chat_id, source_time DESC, message_id DESC);
		CREATE INDEX IF NOT EXISTS idx_messages_observed_at ON messages(observed_at);
		CREATE INDEX IF NOT EXISTS idx_conversations_activity ON conversations(last_activity_at DESC, observed_at DESC, chat_id);
	`); err != nil {
		return fmt.Errorf("ensure account indexes: %w", err)
	}
	return nil
}

func createAccountShard(path string) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return fmt.Errorf("create account shard directory: %w", err)
	}
	db, err := openSQLite(path)
	if err != nil {
		return err
	}
	defer db.Close()
	_, err = db.Exec(`
		CREATE TABLE IF NOT EXISTS schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
		CREATE TABLE IF NOT EXISTS conversations (
			chat_id TEXT PRIMARY KEY,
			chat_type TEXT NOT NULL,
			title TEXT NOT NULL,
			last_activity_at TEXT,
			source TEXT NOT NULL,
			observed_at TEXT NOT NULL,
			directory_state TEXT NOT NULL
		);
		CREATE TABLE IF NOT EXISTS messages (
			message_id TEXT PRIMARY KEY,
			chat_id TEXT NOT NULL,
			source_message_id TEXT NOT NULL,
			direction TEXT NOT NULL,
			message_type TEXT NOT NULL,
			text TEXT NOT NULL,
			source_time TEXT NOT NULL,
			observed_at TEXT NOT NULL,
			source_version TEXT NOT NULL,
			payload_hash TEXT NOT NULL
		);
		CREATE INDEX IF NOT EXISTS idx_messages_chat_time ON messages(chat_id, source_time DESC, message_id DESC);
		CREATE INDEX IF NOT EXISTS idx_messages_observed_at ON messages(observed_at);
		CREATE INDEX IF NOT EXISTS idx_conversations_activity ON conversations(last_activity_at DESC, observed_at DESC, chat_id);
		CREATE TABLE IF NOT EXISTS sync_state (
			stream_key TEXT PRIMARY KEY,
			source_generation TEXT NOT NULL,
			confirmed_sequence INTEGER NOT NULL DEFAULT 0,
			confirmed_cursor TEXT NOT NULL DEFAULT '',
			coverage_state TEXT NOT NULL DEFAULT 'unknown',
			last_success_at TEXT,
			error_code TEXT,
			error_message TEXT,
			updated_at TEXT NOT NULL DEFAULT ''
		);
		CREATE TABLE IF NOT EXISTS ingest_batches (
			batch_id TEXT PRIMARY KEY,
			source_generation TEXT NOT NULL,
			stream_key TEXT NOT NULL,
			sequence INTEGER NOT NULL,
			cursor_start TEXT NOT NULL,
			cursor_end TEXT NOT NULL,
			payload_hash TEXT NOT NULL,
			status TEXT NOT NULL,
			received_at TEXT NOT NULL,
			confirmed_at TEXT
		);
		INSERT INTO schema_meta(key, value) VALUES ('schema_version', '1')
		ON CONFLICT(key) DO UPDATE SET value = excluded.value;
	`)
	if err != nil {
		return fmt.Errorf("initialize account shard: %w", err)
	}
	return nil
}

func openSQLite(path string) (*sql.DB, error) {
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return nil, fmt.Errorf("create sqlite parent directory: %w", err)
	}
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("open sqlite database: %w", err)
	}
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	for _, pragma := range []string{
		"PRAGMA busy_timeout = 5000",
		"PRAGMA journal_mode = WAL",
		"PRAGMA synchronous = FULL",
		"PRAGMA foreign_keys = ON",
	} {
		if _, err := db.Exec(pragma); err != nil {
			_ = db.Close()
			return nil, fmt.Errorf("configure sqlite: %w", err)
		}
	}
	if err := os.Chmod(path, 0o600); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("protect sqlite database: %w", err)
	}
	return db, nil
}

func validateRegistration(registration AccountRegistration) error {
	for name, value := range map[string]string{
		"account id":        registration.AccountID,
		"stable identity":   registration.StableIdentity,
		"source node id":    registration.SourceNodeID,
		"source generation": registration.SourceGeneration,
	} {
		if strings.TrimSpace(value) == "" {
			return fmt.Errorf("%s is required", name)
		}
	}
	if registration.AuthorizationVersion < 0 {
		return errors.New("authorization version must be non-negative")
	}
	return nil
}

func validateScope(scope ReportingScope) error {
	if strings.TrimSpace(scope.ChatID) == "" || strings.TrimSpace(scope.DataType) == "" {
		return errors.New("reporting scope chat id and data type are required")
	}
	if scope.ConfigVersion < 0 {
		return errors.New("scope config version must be non-negative")
	}
	return nil
}

func validateBatch(batch IngestBatch) error {
	if strings.TrimSpace(batch.BatchID) == "" || strings.TrimSpace(batch.SourceGeneration) == "" || strings.TrimSpace(batch.StreamKey) == "" || batch.Sequence < 1 || strings.TrimSpace(batch.PayloadHash) == "" {
		return errors.New("batch id, source generation, stream key, positive sequence, and payload hash are required")
	}
	for _, conversation := range batch.Conversations {
		if strings.TrimSpace(conversation.ChatID) == "" || strings.TrimSpace(conversation.Source) == "" || conversation.ObservedAt.IsZero() {
			return errors.New("conversation identity and observation time are required")
		}
	}
	for _, message := range batch.Messages {
		if strings.TrimSpace(message.MessageID) == "" || strings.TrimSpace(message.ChatID) == "" || strings.TrimSpace(message.PayloadHash) == "" || message.SourceTime.IsZero() || message.ObservedAt.IsZero() {
			return errors.New("message identity, payload hash, source time, and observation time are required")
		}
	}
	return nil
}

func authorizeBatch(batch IngestBatch, scopes []ReportingScope) error {
	if len(scopes) == 0 && (len(batch.Conversations) > 0 || len(batch.Messages) > 0) {
		return ErrAccountNotAuthorized
	}
	for _, conversation := range batch.Conversations {
		if !scopeAllows(scopes, conversation.ChatID, "conversations") {
			return ErrAccountNotAuthorized
		}
	}
	for _, message := range batch.Messages {
		if !scopeAllows(scopes, message.ChatID, "messages") {
			return ErrAccountNotAuthorized
		}
	}
	return nil
}

func scopeAllows(scopes []ReportingScope, chatID, dataType string) bool {
	now := time.Now().UTC()
	for _, scope := range scopes {
		if scope.ChatID != chatID || (scope.ExpiresAt != nil && !scope.ExpiresAt.After(now)) {
			continue
		}
		if scope.DataType == "*" || scope.DataType == "read" || scope.DataType == dataType {
			return true
		}
	}
	return false
}

func accountShardPath(root, accountID string) string {
	digest := sha256.Sum256([]byte(accountID))
	return filepath.Join(root, "accounts", "account-"+hex.EncodeToString(digest[:16]), "data.sqlite")
}

func defaultCoverage(value string) string {
	if value == "" {
		return "complete"
	}
	return value
}

func boolInt(value bool) int {
	if value {
		return 1
	}
	return 0
}

func ptrTime(value time.Time) *time.Time { return &value }

func formatTime(value *time.Time) string {
	if value == nil || value.IsZero() {
		return ""
	}
	return value.UTC().Format(time.RFC3339Nano)
}

func nullableTimeArg(value *time.Time) any {
	if value == nil || value.IsZero() {
		return nil
	}
	return formatTime(value)
}

func accountStorageBytes(path string) (int64, error) {
	var total int64
	for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
		info, err := os.Stat(candidate)
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return 0, err
		}
		total += info.Size()
	}
	return total, nil
}

func estimateBatchBytes(batch IngestBatch) int64 {
	var total int64 = 256
	for _, item := range batch.Conversations {
		total += int64(len(item.ChatID) + len(item.ChatType) + len(item.Title) + len(item.Source) + len(item.DirectoryState) + 128)
	}
	for _, item := range batch.Messages {
		total += int64(len(item.MessageID) + len(item.ChatID) + len(item.SourceMessageID) + len(item.Direction) + len(item.MessageType) + len(item.Text) + len(item.SourceVersion) + len(item.PayloadHash) + 192)
	}
	return total
}

func parseTime(value string) time.Time {
	if value == "" {
		return time.Time{}
	}
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil {
		return time.Time{}
	}
	return parsed
}