package store_test

import (
	"context"
	"fmt"
	"strings"
	"testing"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/stretchr/testify/require"
)

// This is the isolation proof for ADR-012: per-user isolation is enforced by the
// database (migration 003 RLS policies), not merely by application WHERE clauses.
//
// CRITICAL: embedded-postgres's default user (postgres) is a SUPERUSER, which
// BYPASSES RLS regardless of FORCE ROW LEVEL SECURITY. A test that ran scoped
// queries as postgres would be fake-green — it would pass even with the policies
// removed. So this test creates a dedicated NON-SUPERUSER, non-BYPASSRLS role
// ("app", mirroring the production table-owner role tapir which FORCE subjects to
// RLS) and runs every scoped query as that role. The deny-all sanity check below
// (no GUC set → zero rows) proves the enforcement path is live, not bypassed.

// userIsolatedTables are the tables that carry a user_id and whose policy keys
// directly off the tapir.current_user_id GUC. transcripts is deliberately ABSENT
// — ADR-021 made it shared public content (non-RLS); TestTranscriptsTableIsSharedNotRLS
// proves that is the only place the isolation boundary moved.
var userIsolatedTables = []string{
	"users", "videos", "summaries", "summary_actions", "login_events", "video_connections",
}

// allIsolatedTables adds sink_deliveries, whose ownership is derived from its
// summary (no user_id column of its own).
var allIsolatedTables = append(append([]string{}, userIsolatedTables...), "sink_deliveries")

// seeded captures the DB-generated ids for one user's row chain.
type seeded struct {
	userID    string
	videoID   string // videos.id (UUID), reused as summaries.video_id
	summaryID string
}

// seedUser inserts one full chain (user → video → summary → action → delivery)
// as the superuser pool, which bypasses RLS so both users' data lands regardless
// of the GUC. Transcripts are NOT seeded here: they are shared, non-RLS public
// content (ADR-021), so they have no place in a per-user isolation chain.
func seedUser(t *testing.T, p *pgxpool.Pool, userID string) seeded {
	t.Helper()
	ctx := context.Background()

	_, err := p.Exec(ctx, `INSERT INTO users (id) VALUES ($1)`, userID)
	require.NoError(t, err)

	var videoID string
	require.NoError(t, p.QueryRow(ctx,
		`INSERT INTO videos (user_id, provider, provider_video_id, title)
		 VALUES ($1, 'youtube', $2, 'title') RETURNING id`,
		userID, "vid-"+userID).Scan(&videoID))

	var summaryID string
	require.NoError(t, p.QueryRow(ctx,
		`INSERT INTO summaries (user_id, video_id, summary) VALUES ($1, $2, 'sum')
		 RETURNING id`, userID, videoID).Scan(&summaryID))

	_, err = p.Exec(ctx,
		`INSERT INTO summary_actions (user_id, video_id, action)
		 VALUES ($1, $2, 'watched')`, userID, videoID)
	require.NoError(t, err)

	_, err = p.Exec(ctx,
		`INSERT INTO login_events (user_id) VALUES ($1)`, userID)
	require.NoError(t, err)

	_, err = p.Exec(ctx,
		`INSERT INTO sink_deliveries (summary_id, sink, status)
		 VALUES ($1, 'store', 'delivered')`, summaryID)
	require.NoError(t, err)

	_, err = p.Exec(ctx,
		`INSERT INTO video_connections (user_id, provider, token_ref, status)
		 VALUES ($1, 'youtube', $2, 'active')`, userID, "youtube/"+userID+"/refresh_token")
	require.NoError(t, err)

	return seeded{userID: userID, videoID: videoID, summaryID: summaryID}
}

// appPool creates a non-superuser role with DML grants and returns a pool
// connected AS that role, so RLS is actually enforced for it.
func appPool(t *testing.T, super *pgxpool.Pool) *pgxpool.Pool {
	t.Helper()
	ctx := context.Background()

	// Idempotent across tests AND runs: the role persists for the TestMain PG and
	// owns granted privileges, so a plain DROP ROLE fails once any GRANT exists
	// (and more than one test now builds an app pool). Create only if absent; the
	// GRANTs below are themselves idempotent.
	_, err := super.Exec(ctx,
		`DO $$ BEGIN
		    IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'app') THEN
		        CREATE ROLE app LOGIN PASSWORD 'app';
		    END IF;
		 END $$`)
	require.NoError(t, err)
	_, err = super.Exec(ctx, `GRANT USAGE ON SCHEMA public TO app`)
	require.NoError(t, err)
	_, err = super.Exec(ctx,
		`GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app`)
	require.NoError(t, err)

	appDSN := strings.Replace(dsn, "postgres:postgres@", "app:app@", 1)
	p, err := pgxpool.New(ctx, appDSN)
	require.NoError(t, err)
	t.Cleanup(p.Close)

	// Sanity: the app role must NOT be a superuser / must not bypass RLS, else
	// this whole test is theatre.
	var isSuper bool
	require.NoError(t, p.QueryRow(ctx,
		`SELECT rolsuper FROM pg_roles WHERE rolname = current_user`).Scan(&isSuper))
	require.False(t, isSuper, "app role must be non-superuser or RLS is bypassed")
	return p
}

// scopedCount counts rows in table as the app role, optionally scoped to a user
// via the transaction-local GUC. An empty scope sets no GUC (deny-all path).
func scopedCount(t *testing.T, p *pgxpool.Pool, scope, table string) int {
	t.Helper()
	ctx := context.Background()
	tx, err := p.Begin(ctx)
	require.NoError(t, err)
	defer tx.Rollback(ctx) //nolint:errcheck

	if scope != "" {
		_, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, scope)
		require.NoError(t, err)
	}
	var n int
	require.NoError(t, tx.QueryRow(ctx, `SELECT count(*) FROM `+table).Scan(&n))
	return n
}

// scopedRowsAffected runs a write as the app role scoped to scope and returns the
// rows affected, so we can assert a cross-user write touches zero rows.
func scopedRowsAffected(t *testing.T, p *pgxpool.Pool, scope, sql string, args ...any) int64 {
	t.Helper()
	ctx := context.Background()
	tx, err := p.Begin(ctx)
	require.NoError(t, err)
	defer tx.Rollback(ctx) //nolint:errcheck

	_, err = tx.Exec(ctx, `SELECT set_config('tapir.current_user_id', $1, true)`, scope)
	require.NoError(t, err)
	ct, err := tx.Exec(ctx, sql, args...)
	require.NoError(t, err) // RLS hides the rows; it is NOT a permission error
	require.NoError(t, tx.Commit(ctx))
	return ct.RowsAffected()
}

func TestRLSEnforcesPerUserIsolation(t *testing.T) {
	newStore(t) // apply migrations (incl. 003 RLS) as superuser
	super := rawPool(t)
	resetDB(t, super)

	a := seedUser(t, super, userA)
	b := seedUser(t, super, userB)
	app := appPool(t, super)

	// 1. Deny-all: with NO GUC set, every isolated table returns zero rows. This
	//    proves RLS is actually ON (a bypassed/superuser path would see all rows).
	for _, table := range allIsolatedTables {
		require.Equal(t, 0, scopedCount(t, app, "", table),
			"unset tapir.current_user_id must yield deny-all on %s", table)
	}

	// 2. Scoped reads: A sees exactly its own one row per table; likewise B. A
	//    seeing B's row (or vice versa) would mean isolation is broken.
	for _, table := range allIsolatedTables {
		require.Equal(t, 1, scopedCount(t, app, userA, table),
			"user A scoped read must see exactly its own row in %s", table)
		require.Equal(t, 1, scopedCount(t, app, userB, table),
			"user B scoped read must see exactly its own row in %s", table)
	}

	// 3. Cross-user writes are invisible: scoped to A, an UPDATE/DELETE aimed at
	//    B's rows affects zero rows (RLS hides them from the write, too).
	writes := []struct {
		name string
		sql  string
		arg  any // identifies B's row(s)
	}{
		{"update users", `UPDATE users SET display_name = 'hacked' WHERE id = $1`, b.userID},
		{"update videos", `UPDATE videos SET title = 'hacked' WHERE user_id = $1`, b.userID},
		{"queue videos summarize", `UPDATE videos SET summarize_requested = TRUE WHERE id = $1`, b.videoID},
		{"update summaries", `UPDATE summaries SET summary = 'hacked' WHERE user_id = $1`, b.userID},
		{"update summary_actions", `UPDATE summary_actions SET action = 'skipped' WHERE user_id = $1`, b.userID},
		{"update login_events", `UPDATE login_events SET seen_at = NOW() WHERE user_id = $1`, b.userID},
		{"update sink_deliveries", `UPDATE sink_deliveries SET status = 'hacked' WHERE summary_id = $1`, b.summaryID},
		{"update video_connections", `UPDATE video_connections SET token_ref = 'hacked' WHERE user_id = $1`, b.userID},
		{"delete summaries", `DELETE FROM summaries WHERE user_id = $1`, b.userID},
		{"delete summary_actions", `DELETE FROM summary_actions WHERE user_id = $1`, b.userID},
		{"delete login_events", `DELETE FROM login_events WHERE user_id = $1`, b.userID},
		{"delete sink_deliveries", `DELETE FROM sink_deliveries WHERE summary_id = $1`, b.summaryID},
		{"delete video_connections", `DELETE FROM video_connections WHERE user_id = $1`, b.userID},
	}
	for _, w := range writes {
		require.Equal(t, int64(0), scopedRowsAffected(t, app, userA, w.sql, w.arg),
			"user A scoped %s must touch zero of user B's rows", w.name)
	}

	// 4. B's rows survived unchanged (the writes above neither modified nor
	//    deleted them), verified via the superuser pool which bypasses RLS.
	ctx := context.Background()
	var bSummary string
	require.NoError(t, super.QueryRow(ctx,
		`SELECT summary FROM summaries WHERE user_id = $1`, b.userID).Scan(&bSummary))
	require.Equal(t, "sum", bSummary, "B's summary must be untouched by A's writes")

	var bSummaries, bActions, bLogins, bDeliveries, bConnections int
	require.NoError(t, super.QueryRow(ctx,
		`SELECT count(*) FROM summaries WHERE user_id = $1`, b.userID).Scan(&bSummaries))
	require.NoError(t, super.QueryRow(ctx,
		`SELECT count(*) FROM summary_actions WHERE user_id = $1`, b.userID).Scan(&bActions))
	require.NoError(t, super.QueryRow(ctx,
		`SELECT count(*) FROM login_events WHERE user_id = $1`, b.userID).Scan(&bLogins))
	require.NoError(t, super.QueryRow(ctx,
		fmt.Sprintf(`SELECT count(*) FROM sink_deliveries WHERE summary_id = '%s'`, b.summaryID)).Scan(&bDeliveries))
	require.NoError(t, super.QueryRow(ctx,
		`SELECT count(*) FROM video_connections WHERE user_id = $1 AND token_ref <> 'hacked'`, b.userID).Scan(&bConnections))
	require.Equal(t, 1, bSummaries, "A's DELETE must not have removed B's summary")
	require.Equal(t, 1, bActions, "A's DELETE must not have removed B's action")
	require.Equal(t, 1, bLogins, "A's DELETE must not have removed B's login event")
	require.Equal(t, 1, bDeliveries, "A's DELETE must not have removed B's delivery")
	require.Equal(t, 1, bConnections, "A's writes must not have touched B's connection")

	var bRequested bool
	require.NoError(t, super.QueryRow(ctx,
		`SELECT summarize_requested FROM videos WHERE user_id = $1`, b.userID).Scan(&bRequested))
	require.False(t, bRequested, "A scoped must not have queued B's video for summarization")

	_ = a // a's ids are seeded for the symmetric read assertions above
}

// TestTranscriptsTableIsSharedNotRLS is the ADR-021 isolation proof: transcripts
// is the ONE shared, non-RLS surface, and the public-content classification
// leaked to nothing else. It is the inverse of TestRLSEnforcesPerUserIsolation —
// where that asserts deny-all on every user-owned table, this asserts transcripts
// is readable and writable with no user scope at all, holds no user_id, and is
// the single table with row-level security switched off.
func TestTranscriptsTableIsSharedNotRLS(t *testing.T) {
	newStore(t)
	super := rawPool(t)
	resetDB(t, super)
	app := appPool(t, super)
	ctx := context.Background()

	// 1. Shared + non-RLS: with NO GUC set, the app role both writes and reads a
	//    transcript. On an RLS table this would be deny-all (zero rows), exactly as
	//    the main isolation test asserts for every user-owned table.
	_, err := app.Exec(ctx,
		`INSERT INTO transcripts (provider, provider_video_id, source, content)
		 VALUES ('youtube', 'shared-vid', 'captions', 'public words')`)
	require.NoError(t, err, "app role must write shared transcript content with no user scope")
	require.Equal(t, 1, scopedCount(t, app, "", "transcripts"),
		"transcripts must be readable with NO user scope — it is shared, non-RLS (ADR-021)")

	// 2. No user_id column: the table holds only public caption content + the
	//    video's public id, nothing user-identifying.
	var hasUserID bool
	require.NoError(t, super.QueryRow(ctx,
		`SELECT EXISTS (SELECT 1 FROM information_schema.columns
		                 WHERE table_name = 'transcripts' AND column_name = 'user_id')`).Scan(&hasUserID))
	require.False(t, hasUserID, "transcripts must carry no user_id (ADR-021 public content)")

	// 3. The boundary is EXACTLY here: every user-owned table still has row-level
	//    security enabled; transcripts alone has it off. This is the proof the
	//    non-RLS classification was applied to transcripts and leaked nowhere else.
	for _, table := range allIsolatedTables {
		require.True(t, rlsEnabled(t, super, table),
			"%s must still enforce row-level security — isolation must not have regressed", table)
	}
	require.False(t, rlsEnabled(t, super, "transcripts"),
		"transcripts must be the single table with row-level security OFF (the one shared surface)")
}

// rlsEnabled reports whether a public table has ROW LEVEL SECURITY enabled.
func rlsEnabled(t *testing.T, p *pgxpool.Pool, table string) bool {
	t.Helper()
	var enabled bool
	require.NoError(t, p.QueryRow(context.Background(),
		`SELECT relrowsecurity FROM pg_class
		  WHERE relname = $1 AND relnamespace = 'public'::regnamespace`, table).Scan(&enabled))
	return enabled
}