package credentialstore

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
)

// The dialect-specific schemas are idempotent DDL for principal and token tables.
//
// Design notes:
//   - principals.alias is unique so duplicate issuance is rejected at the
//     database layer even under concurrency.
//   - tokens.token_ref is unique and is the only column that can join a
//     verifier (Edge) to a principal. The raw token is never persisted.
//   - tokens.digest is the SHA-256 hex encoding of the raw token; it is the
//     single source of truth for authentication.
//   - tokens.status transitions are enforced by the application layer; the
//     schema only constrains not-null and length.
//   - tokens.revision is an integer monotonic counter used for compare-and-swap
//     disable/revoke. Revoked rows are never re-enabled by any method.
//   - created_at / updated_at / revoked_at are UTC timestamps.
const sqlitePrincipalsSchema = `
CREATE TABLE IF NOT EXISTS principals (
    id         TEXT PRIMARY KEY,
    alias      TEXT NOT NULL UNIQUE,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
`

const sqliteTokensSchema = `
CREATE TABLE IF NOT EXISTS tokens (
    id              TEXT PRIMARY KEY,
    principal_id    TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    token_ref       TEXT NOT NULL UNIQUE,
    digest          TEXT NOT NULL,
    status          TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
    revision        INTEGER NOT NULL DEFAULT 0,
    created_at      TEXT NOT NULL,
    updated_at      TEXT NOT NULL,
    revoked_at      TEXT,
    UNIQUE(principal_id, token_ref)
)
`

const postgresPrincipalsSchema = `
CREATE TABLE IF NOT EXISTS principals (
    id         TEXT PRIMARY KEY,
    alias      TEXT NOT NULL UNIQUE,
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL
)
`

const postgresTokensSchema = `
CREATE TABLE IF NOT EXISTS tokens (
    id              TEXT PRIMARY KEY,
    principal_id    TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    token_ref       TEXT NOT NULL UNIQUE,
    digest          TEXT NOT NULL,
    status          TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
    revision        BIGINT NOT NULL DEFAULT 0,
    created_at      TIMESTAMPTZ NOT NULL,
    updated_at      TIMESTAMPTZ NOT NULL,
    revoked_at      TIMESTAMPTZ,
    UNIQUE(principal_id, token_ref)
)
`

const sqliteCredentialSlotsSchema = `
CREATE TABLE IF NOT EXISTS credential_slots (
    id              TEXT PRIMARY KEY,
    principal_id    TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    vendor          TEXT NOT NULL,
    credential_kind TEXT NOT NULL,
    alias           TEXT,
    status          TEXT NOT NULL CHECK(status IN ('draft','active','disabled','revoked')),
    revision        INTEGER NOT NULL DEFAULT 0,
    algorithm       TEXT NOT NULL,
    key_id          TEXT NOT NULL,
    key_version     INTEGER NOT NULL,
    nonce           BLOB NOT NULL,
    ciphertext      BLOB NOT NULL,
    aad             BLOB,
    created_at      TEXT NOT NULL,
    updated_at      TEXT NOT NULL,
    revoked_at      TEXT,
    UNIQUE(principal_id, id)
)
`

const sqliteSlotRevisionsSchema = `
CREATE TABLE IF NOT EXISTS credential_slot_revisions (
    id          TEXT PRIMARY KEY,
    slot_id     TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
    revision    INTEGER NOT NULL,
    algorithm   TEXT NOT NULL,
    key_id      TEXT NOT NULL,
    key_version INTEGER NOT NULL,
    nonce       BLOB NOT NULL,
    ciphertext  BLOB NOT NULL,
    aad         BLOB,
    created_at  TEXT NOT NULL,
    UNIQUE(slot_id, revision)
)
`

const sqliteRoutesSchema = `
CREATE TABLE IF NOT EXISTS routes (
    id                TEXT PRIMARY KEY,
    principal_id      TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    slot_id           TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
    alias             TEXT,
    profile_id        TEXT NOT NULL,
    upstream_model    TEXT NOT NULL,
    resource_selector TEXT NOT NULL DEFAULT 'default',
    status            TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
    revision          INTEGER NOT NULL DEFAULT 0,
    created_at        TEXT NOT NULL,
    updated_at        TEXT NOT NULL,
    revoked_at        TEXT,
    FOREIGN KEY(principal_id, slot_id) REFERENCES credential_slots(principal_id, id) ON DELETE CASCADE
)
`

const postgresCredentialSlotsSchema = `
CREATE TABLE IF NOT EXISTS credential_slots (
    id              TEXT PRIMARY KEY,
    principal_id    TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    vendor          TEXT NOT NULL,
    credential_kind TEXT NOT NULL,
    alias           TEXT,
    status          TEXT NOT NULL CHECK(status IN ('draft','active','disabled','revoked')),
    revision        BIGINT NOT NULL DEFAULT 0,
    algorithm       TEXT NOT NULL,
    key_id          TEXT NOT NULL,
    key_version     BIGINT NOT NULL,
    nonce           BYTEA NOT NULL,
    ciphertext      BYTEA NOT NULL,
    aad             BYTEA,
    created_at      TIMESTAMPTZ NOT NULL,
    updated_at      TIMESTAMPTZ NOT NULL,
    revoked_at      TIMESTAMPTZ,
    UNIQUE(principal_id, id)
)
`

const postgresSlotRevisionsSchema = `
CREATE TABLE IF NOT EXISTS credential_slot_revisions (
    id          TEXT PRIMARY KEY,
    slot_id     TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
    revision    BIGINT NOT NULL,
    algorithm   TEXT NOT NULL,
    key_id      TEXT NOT NULL,
    key_version BIGINT NOT NULL,
    nonce       BYTEA NOT NULL,
    ciphertext  BYTEA NOT NULL,
    aad         BYTEA,
    created_at  TIMESTAMPTZ NOT NULL,
    UNIQUE(slot_id, revision)
)
`

const postgresRoutesSchema = `
CREATE TABLE IF NOT EXISTS routes (
    id                TEXT PRIMARY KEY,
    principal_id      TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
    slot_id           TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
    alias             TEXT,
    profile_id        TEXT NOT NULL,
    upstream_model    TEXT NOT NULL,
    resource_selector TEXT NOT NULL DEFAULT 'default',
    status            TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
    revision          BIGINT NOT NULL DEFAULT 0,
    created_at        TIMESTAMPTZ NOT NULL,
    updated_at        TIMESTAMPTZ NOT NULL,
    revoked_at        TIMESTAMPTZ,
    FOREIGN KEY(principal_id, slot_id) REFERENCES credential_slots(principal_id, id) ON DELETE CASCADE
)
`

const sqliteProjectionStateSchema = `
CREATE TABLE IF NOT EXISTS principal_projection_state (
    singleton  INTEGER PRIMARY KEY CHECK(singleton = 1),
    generation INTEGER NOT NULL CHECK(generation >= 0)
)
`

const postgresProjectionStateSchema = `
CREATE TABLE IF NOT EXISTS principal_projection_state (
    singleton  SMALLINT PRIMARY KEY CHECK(singleton = 1),
    generation BIGINT NOT NULL CHECK(generation >= 0)
)
`

// migrate runs the principal/token/slot/route DDL and any post-migration adjustments.
// It is safe to call repeatedly; every statement is idempotent.
func migrate(ctx context.Context, db *sql.DB, dialect string) error {
	for _, statement := range schemaStatements(dialect) {
		if _, err := db.ExecContext(ctx, statement); err != nil {
			return fmt.Errorf("schema: %w", err)
		}
	}
	if err := migrateAddRevokedAt(ctx, db, dialect); err != nil {
		return fmt.Errorf("migrate add revoked_at: %w", err)
	}
	if err := migrateCredentialSlotContract(ctx, db, dialect); err != nil {
		return fmt.Errorf("migrate credential slot contract: %w", err)
	}
	if err := validateMigratedCredentialBindings(ctx, db, dialect); err != nil {
		return fmt.Errorf("validate migrated credential bindings: %w", err)
	}
	if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_tokens_principal_token ON tokens(principal_id, token_ref)`); err != nil {
		return fmt.Errorf("migrate token_ref unique: %w", err)
	}
	if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_routes_principal_alias ON routes(principal_id, alias) WHERE alias IS NOT NULL AND alias != ''`); err != nil {
		return fmt.Errorf("migrate route alias unique: %w", err)
	}
	if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_credential_slots_principal_alias ON credential_slots(principal_id, alias) WHERE alias IS NOT NULL AND alias != ''`); err != nil {
		return fmt.Errorf("migrate slot alias unique: %w", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO principal_projection_state (singleton, generation) VALUES (1, 0) ON CONFLICT(singleton) DO NOTHING`); err != nil {
		return fmt.Errorf("migrate projection generation: %w", err)
	}
	return nil
}

func schemaStatements(dialect string) []string {
	principals, tokens := sqlitePrincipalsSchema, sqliteTokensSchema
	slots, revisions, routes := sqliteCredentialSlotsSchema, sqliteSlotRevisionsSchema, sqliteRoutesSchema
	projectionState := sqliteProjectionStateSchema
	if dialect == dialectPostgres {
		principals, tokens = postgresPrincipalsSchema, postgresTokensSchema
		slots, revisions, routes = postgresCredentialSlotsSchema, postgresSlotRevisionsSchema, postgresRoutesSchema
		projectionState = postgresProjectionStateSchema
	}
	return []string{
		principals,
		tokens,
		slots,
		revisions,
		routes,
		projectionState,
		`CREATE INDEX IF NOT EXISTS idx_tokens_digest ON tokens(digest)`,
		`CREATE INDEX IF NOT EXISTS idx_tokens_principal_status ON tokens(principal_id, status)`,
		`CREATE INDEX IF NOT EXISTS idx_credential_slots_principal_status ON credential_slots(principal_id, status)`,
		`CREATE INDEX IF NOT EXISTS idx_routes_principal_status ON routes(principal_id, status)`,
		`CREATE INDEX IF NOT EXISTS idx_routes_slot_id ON routes(slot_id)`,
	}
}

// migrateAddRevokedAt adds the revoked_at column if missing. The original
// schema did not include it; this migration backfills it for existing rows.
func migrateAddRevokedAt(ctx context.Context, db *sql.DB, dialect string) error {
	if dialect == dialectPostgres {
		_, err := db.ExecContext(ctx, `ALTER TABLE tokens ADD COLUMN IF NOT EXISTS revoked_at TIMESTAMPTZ`)
		return err
	}
	if hasColumn(ctx, db, dialect, "tokens", "revoked_at") {
		return nil
	}
	_, err := db.ExecContext(ctx, `ALTER TABLE tokens ADD COLUMN revoked_at TEXT`)
	return err
}

// migrateCredentialSlotContract upgrades the first credential catalog schema
// to the S04 contract. Existing rows receive the deterministic bearer kind,
// while new rows can remain draft and omit an alias. SQLite needs a table
// rebuild because it cannot relax NOT NULL or replace a CHECK constraint.
func migrateCredentialSlotContract(ctx context.Context, db *sql.DB, dialect string) error {
	if dialect == dialectPostgres {
		hadCredentialKind := hasColumn(ctx, db, dialect, "credential_slots", "credential_kind")
		if !hadCredentialKind {
			if _, err := db.ExecContext(ctx, `ALTER TABLE credential_slots ADD COLUMN credential_kind TEXT NOT NULL DEFAULT 'bearer'`); err != nil {
				return err
			}
			if _, err := db.ExecContext(ctx, `UPDATE credential_slots
				SET credential_kind = CASE
					WHEN lower(trim(vendor)) IN ('anthropic', 'mimo_messages', 'seulgi_messages', 'seulgivibe_claude') THEN 'api_key'
					ELSE 'bearer'
				END`); err != nil {
				return err
			}
		}
		if _, err := db.ExecContext(ctx, `ALTER TABLE credential_slots ALTER COLUMN alias DROP NOT NULL`); err != nil {
			return err
		}
		_, err := db.ExecContext(ctx, `
DO $$
BEGIN
    IF EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'credential_slots_status_check') THEN
        ALTER TABLE credential_slots DROP CONSTRAINT credential_slots_status_check;
    END IF;
    IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'credential_slots_status_check') THEN
        ALTER TABLE credential_slots ADD CONSTRAINT credential_slots_status_check
            CHECK (status IN ('draft','active','disabled','revoked'));
    END IF;
END $$`)
		return err
	}

	if hasColumn(ctx, db, dialect, "credential_slots", "credential_kind") && sqliteSlotSchemaIsCurrent(ctx, db) {
		return nil
	}
	return rebuildSQLiteCredentialCatalog(ctx, db)
}

func sqliteSlotSchemaIsCurrent(ctx context.Context, db *sql.DB) bool {
	var sqlText string
	if err := db.QueryRowContext(ctx, `SELECT sql FROM sqlite_master WHERE type='table' AND name='credential_slots'`).Scan(&sqlText); err != nil {
		return false
	}
	return strings.Contains(strings.ToLower(sqlText), "'draft'")
}

func rebuildSQLiteCredentialCatalog(ctx context.Context, db *sql.DB) (err error) {
	conn, err := db.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	if _, err := conn.ExecContext(ctx, `PRAGMA foreign_keys = OFF`); err != nil {
		return err
	}
	defer func() {
		_, restoreErr := conn.ExecContext(ctx, `PRAGMA foreign_keys = ON`)
		if err == nil && restoreErr != nil {
			err = restoreErr
		}
	}()

	tx, err := conn.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()
	for _, statement := range []string{
		`ALTER TABLE routes RENAME TO routes_legacy_credential_catalog`,
		`ALTER TABLE credential_slot_revisions RENAME TO credential_slot_revisions_legacy_credential_catalog`,
		`ALTER TABLE credential_slots RENAME TO credential_slots_legacy_credential_catalog`,
		sqliteCredentialSlotsSchema,
		sqliteSlotRevisionsSchema,
		sqliteRoutesSchema,
		`INSERT INTO credential_slots (id, principal_id, vendor, credential_kind, alias, status, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at, updated_at, revoked_at)
		 SELECT id, principal_id, vendor,
			CASE WHEN lower(trim(vendor)) IN ('anthropic', 'mimo_messages', 'seulgi_messages', 'seulgivibe_claude') THEN 'api_key' ELSE 'bearer' END,
			alias, status, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at, updated_at, revoked_at
		 FROM credential_slots_legacy_credential_catalog`,
		`INSERT INTO credential_slot_revisions (id, slot_id, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at)
		 SELECT id, slot_id, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at
		 FROM credential_slot_revisions_legacy_credential_catalog`,
		`INSERT INTO routes (id, principal_id, slot_id, alias, profile_id, upstream_model, resource_selector, status, revision, created_at, updated_at, revoked_at)
		 SELECT id, principal_id, slot_id, alias, profile_id, upstream_model, resource_selector, status, revision, created_at, updated_at, revoked_at
		 FROM routes_legacy_credential_catalog`,
		`DROP TABLE routes_legacy_credential_catalog`,
		`DROP TABLE credential_slot_revisions_legacy_credential_catalog`,
		`DROP TABLE credential_slots_legacy_credential_catalog`,
	} {
		if _, err = tx.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	err = tx.Commit()
	return err
}

// validateMigratedCredentialBindings fences legacy active state using the
// same closed resolver used by route mutations. It deliberately reports only
// typed, redacted compatibility failures: database rows can contain encrypted
// credential material but no migration error may expose it.
func validateMigratedCredentialBindings(ctx context.Context, db *sql.DB, dialect string) error {
	activeRoutes, err := db.QueryContext(ctx, bindQuery(dialect, `
		SELECT r.principal_id, r.slot_id, r.profile_id,
		       s.id, s.principal_id, s.vendor, s.credential_kind, s.alias, s.status
		FROM routes r
		LEFT JOIN credential_slots s ON s.id = r.slot_id
		WHERE r.status = ?
	`), StatusActive)
	if err != nil {
		return err
	}
	defer activeRoutes.Close()

	for activeRoutes.Next() {
		var routePrincipalID string
		var routeSlotID string
		var profileID string
		var joinedSlotID, joinedPrincipalID, joinedVendor, joinedKind, slotAlias, joinedStatus sql.NullString
		if err := activeRoutes.Scan(
			&routePrincipalID, &routeSlotID, &profileID,
			&joinedSlotID, &joinedPrincipalID, &joinedVendor, &joinedKind, &slotAlias, &joinedStatus,
		); err != nil {
			return err
		}
		if !joinedSlotID.Valid {
			return fmt.Errorf("%w: active migrated route has no slot", ErrIncompatibleProfile)
		}
		slot := CredentialSlotRecord{
			ID:             joinedSlotID.String,
			PrincipalID:    joinedPrincipalID.String,
			Vendor:         joinedVendor.String,
			CredentialKind: joinedKind.String,
			Status:         joinedStatus.String,
		}
		if slotAlias.Valid {
			slot.Alias = slotAlias.String
		}
		if routePrincipalID != slot.PrincipalID {
			return fmt.Errorf("%w: active migrated route has mismatched ownership", ErrIncompatibleProfile)
		}
		switch slot.Status {
		case StatusActive, StatusDisabled, StatusRevoked:
			// Disabled and revoked slots retain their bindings so a disabled slot
			// can be re-enabled without recreating routes. They are not serving-ready,
			// but their retained active bindings must still be compatible below.
		default:
			return fmt.Errorf("%w: active migrated route has an unavailable slot", ErrIncompatibleProfile)
		}
		if _, err := resolveSlotProfile(slot, profileID); err != nil {
			return fmt.Errorf("active migrated route compatibility: %w", err)
		}
	}
	if err := activeRoutes.Err(); err != nil {
		return err
	}

	activeSlots, err := db.QueryContext(ctx, bindQuery(dialect, `
		SELECT id, principal_id, vendor, credential_kind, alias, status
		FROM credential_slots
		WHERE status = ?
	`), StatusActive)
	if err != nil {
		return err
	}
	defer activeSlots.Close()

	for activeSlots.Next() {
		var slot CredentialSlotRecord
		var slotAlias sql.NullString
		if err := activeSlots.Scan(&slot.ID, &slot.PrincipalID, &slot.Vendor, &slot.CredentialKind, &slotAlias, &slot.Status); err != nil {
			return err
		}
		if slotAlias.Valid {
			slot.Alias = slotAlias.String
		}
		if err := migratedSlotHasCompatibleBinding(ctx, db, dialect, slot); err != nil {
			return err
		}
	}
	return activeSlots.Err()
}

func migratedSlotHasCompatibleBinding(ctx context.Context, db *sql.DB, dialect string, slot CredentialSlotRecord) error {
	rows, err := db.QueryContext(ctx, bindQuery(dialect, `
		SELECT profile_id
		FROM routes
		WHERE principal_id = ? AND slot_id = ? AND status <> ?
	`), slot.PrincipalID, slot.ID, StatusRevoked)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var profileID string
		if err := rows.Scan(&profileID); err != nil {
			return err
		}
		if _, err := resolveSlotProfile(slot, profileID); err == nil {
			return nil
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	return fmt.Errorf("%w: active migrated slot has no compatible binding", ErrIncompatibleProfile)
}

// hasColumn reports whether table has a column named col.
// Uses PRAGMA table_info for SQLite compatibility and information_schema for PostgreSQL.
func hasColumn(ctx context.Context, db *sql.DB, dialect, table, col string) bool {
	if dialect == dialectPostgres {
		var name string
		err := db.QueryRowContext(ctx,
			`SELECT column_name FROM information_schema.columns WHERE table_name=$1 AND column_name=$2`, table, col,
		).Scan(&name)
		return err == nil
	}
	rows, err := db.QueryContext(ctx, `PRAGMA table_info(`+table+`)`)
	if err == nil {
		defer rows.Close()
		for rows.Next() {
			var cid int
			var name, dtype string
			var notnull, pk int
			var dfltValue interface{}
			if err := rows.Scan(&cid, &name, &dtype, &notnull, &dfltValue, &pk); err != nil {
				return false
			}
			if name == col {
				return true
			}
		}
		return false
	}
	return false
}