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.
var userIsolatedTables = []string{
	"users", "videos", "transcripts", "summaries", "summary_actions",
}

// 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 → transcript → summary →
// action → delivery) as the superuser pool, which bypasses RLS so both users'
// data lands regardless of the GUC.
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))

	_, err = p.Exec(ctx,
		`INSERT INTO transcripts (video_id, user_id, source, content)
		 VALUES ($1, $2, 'captions', 'words')`, videoID, userID)
	require.NoError(t, err)

	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 sink_deliveries (summary_id, sink, status)
		 VALUES ($1, 'store', 'delivered')`, summaryID)
	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 test runs (schema/role persist for the TestMain PG).
	_, _ = super.Exec(ctx, `DROP ROLE IF EXISTS app`)
	_, err := super.Exec(ctx, `CREATE ROLE app LOGIN PASSWORD 'app'`)
	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},
		{"update transcripts", `UPDATE transcripts SET content = 'hacked' WHERE user_id = $1`, b.userID},
		{"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 sink_deliveries", `UPDATE sink_deliveries SET status = 'hacked' WHERE summary_id = $1`, b.summaryID},
		{"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 sink_deliveries", `DELETE FROM sink_deliveries WHERE summary_id = $1`, b.summaryID},
	}
	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, bDeliveries 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,
		fmt.Sprintf(`SELECT count(*) FROM sink_deliveries WHERE summary_id = '%s'`, b.summaryID)).Scan(&bDeliveries))
	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, bDeliveries, "A's DELETE must not have removed B's delivery")

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