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", "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 → 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)

	_, 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 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},
		{"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 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, 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,
		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, 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")

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