user_cleanup_postgres_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446
  1. package service
  2. import (
  3. "context"
  4. "database/sql"
  5. "fmt"
  6. "net/url"
  7. "os"
  8. "strings"
  9. "sync"
  10. "testing"
  11. "time"
  12. "github.com/Wei-Shaw/sub2api/migrations"
  13. "github.com/google/uuid"
  14. "github.com/stretchr/testify/require"
  15. )
  16. // This suite never targets the deployed database. Supply an isolated database
  17. // whose name ends in _test; every case uses its own temporary, random schema.
  18. func cleanupTestDB(t *testing.T) (*sql.DB, *UserCleanupService, int64) {
  19. t.Helper()
  20. dsn := os.Getenv("USER_CLEANUP_TEST_DSN")
  21. if dsn == "" {
  22. if os.Getenv("REQUIRE_DATABASE_TESTS") == "1" {
  23. t.Fatal("USER_CLEANUP_TEST_DSN is required in the database CI job")
  24. }
  25. t.Skip("USER_CLEANUP_TEST_DSN is not set; real PostgreSQL cleanup tests require an isolated *_test database")
  26. }
  27. if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") {
  28. parsed, err := url.Parse(dsn)
  29. require.NoError(t, err)
  30. q := parsed.Query()
  31. q.Set("connect_timeout", "10")
  32. parsed.RawQuery = q.Encode()
  33. dsn = parsed.String()
  34. } else {
  35. dsn += " connect_timeout=10"
  36. }
  37. base, err := sql.Open("postgres", dsn)
  38. require.NoError(t, err)
  39. var database string
  40. require.NoError(t, base.QueryRow(`SELECT current_database()`).Scan(&database))
  41. require.True(t, strings.HasSuffix(database, "_test"), "refusing any database without the _test suffix")
  42. schema := "cleanup_test_" + strings.ReplaceAll(uuid.NewString(), "-", "")
  43. _, err = base.Exec(`CREATE SCHEMA ` + schema)
  44. require.NoError(t, err)
  45. if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") {
  46. parsed, err := url.Parse(dsn)
  47. require.NoError(t, err)
  48. q := parsed.Query()
  49. q.Set("search_path", schema)
  50. parsed.RawQuery = q.Encode()
  51. dsn = parsed.String()
  52. } else {
  53. dsn += " search_path=" + schema
  54. }
  55. db, err := sql.Open("postgres", dsn)
  56. require.NoError(t, err)
  57. db.SetMaxOpenConns(10)
  58. t.Cleanup(func() {
  59. _ = db.Close()
  60. cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  61. defer cancel()
  62. _, cleanupErr := base.ExecContext(cleanupCtx, `DROP SCHEMA `+schema+` CASCADE`)
  63. if cleanupErr != nil {
  64. t.Errorf("remove isolated cleanup schema: %v", cleanupErr)
  65. }
  66. _ = base.Close()
  67. })
  68. _, err = db.Exec(cleanupTestSchema)
  69. require.NoError(t, err)
  70. for _, name := range []string{"245_user_cleanup.sql", "245_user_cleanup.sql"} {
  71. migration, err := migrations.FS.ReadFile(name)
  72. require.NoError(t, err)
  73. _, err = db.Exec(string(migration))
  74. require.NoError(t, err)
  75. }
  76. var actor int64
  77. require.NoError(t, db.QueryRow(`INSERT INTO users(email,role) VALUES ('super@example.test','super_admin') RETURNING id`).Scan(&actor))
  78. return db, NewUserCleanupService(db, nil), actor
  79. }
  80. const cleanupTestSchema = `
  81. CREATE TABLE users(id BIGSERIAL PRIMARY KEY,email TEXT NOT NULL DEFAULT '',username TEXT NOT NULL DEFAULT '',role TEXT NOT NULL DEFAULT 'user',status TEXT NOT NULL DEFAULT 'active',balance NUMERIC(20,8) NOT NULL DEFAULT 0,frozen_balance NUMERIC(20,8) NOT NULL DEFAULT 0,created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()-INTERVAL '7 days',updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),last_login_at TIMESTAMPTZ,last_active_at TIMESTAMPTZ,deleted_at TIMESTAMPTZ);
  82. CREATE UNIQUE INDEX cleanup_user_email_active ON users(email) WHERE deleted_at IS NULL AND email<>'';
  83. CREATE TABLE usage_logs(id BIGSERIAL PRIMARY KEY,user_id BIGINT REFERENCES users(id),created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()-INTERVAL '2 days');
  84. CREATE TABLE api_keys(id BIGSERIAL PRIMARY KEY,user_id BIGINT REFERENCES users(id),key TEXT NOT NULL UNIQUE,deleted_at TIMESTAMPTZ,last_used_at TIMESTAMPTZ,updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW());
  85. CREATE TABLE user_subscriptions(user_id BIGINT REFERENCES users(id),expires_at TIMESTAMPTZ,status TEXT,deleted_at TIMESTAMPTZ);
  86. CREATE TABLE payment_orders(user_id BIGINT,expires_at TIMESTAMPTZ,status TEXT);
  87. CREATE TABLE batch_image_jobs(user_id BIGINT,status TEXT,hold_amount NUMERIC(20,8),settled_at TIMESTAMPTZ);
  88. CREATE TABLE audit_logs(id BIGSERIAL PRIMARY KEY,actor_user_id BIGINT,actor_email TEXT,actor_role TEXT,auth_method TEXT,action TEXT,method TEXT,path TEXT,request_id TEXT,client_ip TEXT,status_code INT,extra JSONB);
  89. CREATE TABLE auth_identities(id BIGSERIAL PRIMARY KEY,user_id BIGINT REFERENCES users(id),provider_type TEXT,provider_key TEXT,provider_subject TEXT,metadata JSONB NOT NULL DEFAULT '{}'::jsonb,UNIQUE(provider_type,provider_key,provider_subject));
  90. CREATE TABLE auth_identity_channels(id BIGSERIAL PRIMARY KEY,identity_id BIGINT REFERENCES auth_identities(id) ON DELETE CASCADE,channel_subject TEXT UNIQUE);
  91. CREATE TABLE identity_adoption_decisions(id BIGSERIAL PRIMARY KEY,identity_id BIGINT REFERENCES auth_identities(id) ON DELETE SET NULL);
  92. `
  93. func cleanupTestUser(t *testing.T, db *sql.DB, balance string) int64 {
  94. t.Helper()
  95. var id int64
  96. require.NoError(t, db.QueryRow(`INSERT INTO users(email,balance) VALUES ($1,$2) RETURNING id`, uuid.NewString()+"@example.test", balance).Scan(&id))
  97. _, err := db.Exec(`INSERT INTO usage_logs(user_id) VALUES ($1)`, id)
  98. require.NoError(t, err)
  99. _, err = db.Exec(`INSERT INTO api_keys(user_id,key) VALUES ($1,$2)`, id, "test-key-"+uuid.NewString())
  100. require.NoError(t, err)
  101. _, err = db.Exec(`INSERT INTO auth_identities(user_id,provider_type,provider_key,provider_subject) SELECT id,'email','email',email FROM users WHERE id=$1`, id)
  102. require.NoError(t, err)
  103. return id
  104. }
  105. func TestUserCleanupPostgresBoundaries(t *testing.T) {
  106. db, svc, actor := cleanupTestDB(t)
  107. cases := []struct {
  108. name, balance, mutation string
  109. eligible bool
  110. }{
  111. {"zero", "0", "", true}, {"negative", "-0.00000001", "", true}, {"positive", "0.00000001", "", false},
  112. {"recent_usage", "0", `UPDATE usage_logs SET created_at=NOW() WHERE user_id=$1`, false},
  113. {"future_usage", "0", `UPDATE usage_logs SET created_at=NOW()+INTERVAL '1 day' WHERE user_id=$1`, false},
  114. {"missing_history_old", "0", `DELETE FROM usage_logs WHERE user_id=$1`, false},
  115. {"recent_registration", "0", `UPDATE users SET created_at=NOW() WHERE id=$1`, false},
  116. {"admin", "0", `UPDATE users SET role='admin' WHERE id=$1`, false},
  117. {"super_admin", "0", `UPDATE users SET role='super_admin' WHERE id=$1`, false},
  118. {"system", "0", `INSERT INTO user_cleanup_guards(user_id,is_system) VALUES ($1,TRUE)`, false},
  119. {"protected", "0", `INSERT INTO user_cleanup_guards(user_id,is_protected) VALUES ($1,TRUE)`, false},
  120. {"frozen_assets", "0", `UPDATE users SET frozen_balance=1 WHERE id=$1`, false},
  121. {"active_subscription", "0", `INSERT INTO user_subscriptions(user_id,status,expires_at) VALUES ($1,'active',NOW()+INTERVAL '1 day')`, false},
  122. {"suspended_subscription", "0", `INSERT INTO user_subscriptions(user_id,status,expires_at) VALUES ($1,'suspended',NOW()+INTERVAL '1 day')`, false},
  123. {"expired_subscription", "0", `INSERT INTO user_subscriptions(user_id,status,expires_at) VALUES ($1,'active',NOW()-INTERVAL '1 day')`, true},
  124. {"pending_payment", "0", `INSERT INTO payment_orders(user_id,status,expires_at) VALUES ($1,'PENDING',NOW()+INTERVAL '1 day')`, false},
  125. {"recharging", "0", `INSERT INTO payment_orders(user_id,status,expires_at) VALUES ($1,'RECHARGING',NOW()-INTERVAL '1 day')`, false},
  126. {"historical_payment", "0", `INSERT INTO payment_orders(user_id,status,expires_at) VALUES ($1,'COMPLETED',NOW()-INTERVAL '1 day')`, true},
  127. {"expired_payment", "0", `INSERT INTO payment_orders(user_id,status,expires_at) VALUES ($1,'PENDING',NOW()-INTERVAL '1 day')`, true},
  128. {"batch_running", "0", `INSERT INTO batch_image_jobs(user_id,status) VALUES ($1,'running')`, false},
  129. {"batch_unsettled", "0", `INSERT INTO batch_image_jobs(user_id,status,hold_amount) VALUES ($1,'failed',1)`, false},
  130. {"recent_login", "0", `UPDATE users SET last_login_at=NOW() WHERE id=$1`, false},
  131. {"recent_activity", "0", `UPDATE users SET last_active_at=NOW() WHERE id=$1`, false},
  132. {"recent_api_key", "0", `UPDATE api_keys SET last_used_at=NOW() WHERE user_id=$1`, false},
  133. }
  134. want := map[int64]bool{}
  135. for _, tc := range cases {
  136. id := cleanupTestUser(t, db, tc.balance)
  137. if tc.mutation != "" {
  138. _, err := db.Exec(tc.mutation, id)
  139. require.NoError(t, err, tc.name)
  140. }
  141. want[id] = tc.eligible
  142. }
  143. preview, err := svc.Preview(context.Background(), actor)
  144. require.NoError(t, err)
  145. got := map[int64]bool{}
  146. for _, candidate := range preview.Candidates {
  147. got[candidate.ID] = true
  148. }
  149. for id, eligible := range want {
  150. require.Equal(t, eligible, got[id], "user %d", id)
  151. }
  152. require.Equal(t, 5, preview.Total)
  153. require.Equal(t, 4, preview.ZeroBalance)
  154. require.Equal(t, 1, preview.NegativeBalance)
  155. // NOW() remains fixed within this transaction, so the exact 24-hour boundary
  156. // is tested without clock drift between the fixture write and the predicate.
  157. tx, err := db.Begin()
  158. require.NoError(t, err)
  159. defer func() { _ = tx.Rollback() }()
  160. _, err = tx.Exec(`UPDATE usage_logs SET created_at=NOW()-INTERVAL '24 hours'`)
  161. require.NoError(t, err)
  162. items, err := cleanupCandidates(context.Background(), tx, nil)
  163. require.NoError(t, err)
  164. require.Empty(t, items)
  165. }
  166. func TestUserCleanupPostgresConcurrentRevalidationAndReplay(t *testing.T) {
  167. db, svc, actor := cleanupTestDB(t)
  168. ctx := context.Background()
  169. credit := cleanupTestUser(t, db, "0")
  170. used := cleanupTestUser(t, db, "0")
  171. removed := cleanupTestUser(t, db, "-1")
  172. preview, err := svc.Preview(ctx, actor)
  173. require.NoError(t, err)
  174. tx, err := db.Begin()
  175. require.NoError(t, err)
  176. defer func() { _ = tx.Rollback() }()
  177. _, err = tx.Exec(`UPDATE users SET balance=5 WHERE id=$1`, credit)
  178. require.NoError(t, err)
  179. // The insert holds PostgreSQL's FK KEY SHARE user lock until commit.
  180. _, err = tx.Exec(`INSERT INTO usage_logs(user_id,created_at) VALUES ($1,NOW())`, used)
  181. require.NoError(t, err)
  182. done := make(chan *UserCleanupResult, 1)
  183. failed := make(chan error, 1)
  184. go func() {
  185. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 3, UserCleanupAuditContext{})
  186. if err != nil {
  187. failed <- err
  188. return
  189. }
  190. done <- result
  191. }()
  192. // Observe an actual lock waiter, rather than treating elapsed time as evidence.
  193. require.Eventually(t, func() bool {
  194. var waiting bool
  195. err := db.QueryRow(`SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE 'SELECT id FROM users WHERE id=ANY%')`).Scan(&waiting)
  196. return err == nil && waiting
  197. }, 10*time.Second, 50*time.Millisecond)
  198. require.NoError(t, tx.Commit())
  199. var result *UserCleanupResult
  200. select {
  201. case result = <-done:
  202. case err := <-failed:
  203. require.NoError(t, err)
  204. case <-time.After(20 * time.Second):
  205. t.Fatal("cleanup did not finish")
  206. }
  207. require.Equal(t, []int64{removed}, result.UserIDs)
  208. require.Equal(t, 2, result.SkippedCount)
  209. require.Equal(t, 1, result.NegativeBalance)
  210. var deleted bool
  211. var key string
  212. require.NoError(t, db.QueryRow(`SELECT deleted_at IS NOT NULL AND status='disabled' FROM users WHERE id=$1`, removed).Scan(&deleted))
  213. require.True(t, deleted)
  214. require.NoError(t, db.QueryRow(`SELECT key FROM api_keys WHERE user_id=$1`, removed).Scan(&key))
  215. require.True(t, strings.HasPrefix(key, "__deleted__cleanup__"))
  216. // Two retries after a lost response both retrieve the persisted original result.
  217. var wg sync.WaitGroup
  218. for range 2 {
  219. wg.Add(1)
  220. go func() {
  221. defer wg.Done()
  222. replay, err := svc.Execute(ctx, actor, preview.PreviewID, true, 3, UserCleanupAuditContext{})
  223. if err != nil {
  224. failed <- err
  225. return
  226. }
  227. if !replay.Replayed || replay.DeletedCount != 1 {
  228. failed <- fmt.Errorf("invalid replay: %+v", replay)
  229. }
  230. }()
  231. }
  232. wg.Wait()
  233. close(failed)
  234. for err := range failed {
  235. require.NoError(t, err)
  236. }
  237. var audits, users, usage int
  238. require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM audit_logs WHERE action='admin.users.cleanup' AND extra->'user_ids'=$1::jsonb`, fmt.Sprintf("[%d]", removed)).Scan(&audits))
  239. require.Equal(t, 1, audits)
  240. require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM users`).Scan(&users))
  241. require.Equal(t, 4, users, "no physical user deletion")
  242. require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM usage_logs`).Scan(&usage))
  243. require.Equal(t, 4, usage, "history retained")
  244. }
  245. func TestUserCleanupPostgresAuditRollbackAndPermissions(t *testing.T) {
  246. db, svc, actor := cleanupTestDB(t)
  247. ctx := context.Background()
  248. user := cleanupTestUser(t, db, "0")
  249. cleanupTestIdentityRelations(t, db, user)
  250. preview, err := svc.Preview(ctx, actor)
  251. require.NoError(t, err)
  252. _, err = svc.Execute(ctx, user, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  253. require.ErrorIs(t, err, ErrUserCleanupForbidden)
  254. var other int64
  255. require.NoError(t, db.QueryRow(`INSERT INTO users(role) VALUES ('admin') RETURNING id`).Scan(&other))
  256. _, err = svc.Execute(ctx, other, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  257. require.ErrorIs(t, err, ErrUserCleanupPreview)
  258. _, err = svc.Execute(ctx, actor, preview.PreviewID, true, 2, UserCleanupAuditContext{})
  259. require.ErrorIs(t, err, ErrUserCleanupConfirmation)
  260. _, err = db.Exec(`ALTER TABLE audit_logs ADD CONSTRAINT reject_cleanup CHECK(action<>'admin.users.cleanup')`)
  261. require.NoError(t, err)
  262. _, err = svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  263. require.Error(t, err)
  264. var intact bool
  265. require.NoError(t, db.QueryRow(`SELECT u.deleted_at IS NULL AND k.deleted_at IS NULL AND k.key NOT LIKE '__deleted__%' FROM users u JOIN api_keys k ON k.user_id=u.id WHERE u.id=$1`, user).Scan(&intact))
  266. require.True(t, intact, "audit failure must roll back both user and key mutation")
  267. var identityIntact bool
  268. require.NoError(t, db.QueryRow(`SELECT EXISTS(SELECT 1 FROM auth_identities WHERE user_id=$1) AND NOT EXISTS(SELECT 1 FROM user_cleanup_identity_archives WHERE user_id=$1)`, user).Scan(&identityIntact))
  269. require.True(t, identityIntact, "audit failure must roll back identity archiving and unbinding")
  270. require.NoError(t, db.QueryRow(`SELECT EXISTS(SELECT 1 FROM auth_identity_channels c JOIN auth_identities a ON a.id=c.identity_id WHERE a.user_id=$1) AND EXISTS(SELECT 1 FROM identity_adoption_decisions d JOIN auth_identities a ON a.id=d.identity_id WHERE a.user_id=$1)`, user).Scan(&identityIntact))
  271. require.True(t, identityIntact, "audit failure must also roll back channel deletion and decision detachment")
  272. _, err = db.Exec(`ALTER TABLE audit_logs DROP CONSTRAINT reject_cleanup`)
  273. require.NoError(t, err)
  274. system := true
  275. _, err = svc.UpdateGuard(ctx, other, user, false, &system, UserCleanupAuditContext{})
  276. require.Error(t, err, "admin cannot mark system accounts")
  277. _, err = svc.UpdateGuard(ctx, actor, user, false, &system, UserCleanupAuditContext{})
  278. require.NoError(t, err)
  279. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  280. require.NoError(t, err)
  281. require.Zero(t, result.DeletedCount)
  282. require.Equal(t, 1, result.SkippedCount)
  283. _, err = svc.UpdateGuard(ctx, actor, other, true, nil, UserCleanupAuditContext{})
  284. require.Error(t, err, "administrators are inherently protected")
  285. }
  286. func TestUserCleanupRequiresExplicitConfirmation(t *testing.T) {
  287. svc := NewUserCleanupService(nil, nil)
  288. for _, count := range []int{-1, 0, 501} {
  289. _, err := svc.Execute(context.Background(), 1, uuid.NewString(), true, count, UserCleanupAuditContext{})
  290. require.ErrorIs(t, err, ErrUserCleanupConfirmation)
  291. }
  292. _, err := svc.Execute(context.Background(), 1, uuid.NewString(), false, 1, UserCleanupAuditContext{})
  293. require.ErrorIs(t, err, ErrUserCleanupConfirmation)
  294. _, err = svc.Execute(context.Background(), 1, "invalid", true, 1, UserCleanupAuditContext{})
  295. require.ErrorIs(t, err, ErrUserCleanupPreview)
  296. }
  297. func TestUserCleanupPostgresConcurrentActivityLocks(t *testing.T) {
  298. for _, tc := range []struct{ name, mutation, waitingQuery string }{
  299. {"usage_fk", `INSERT INTO usage_logs(user_id,created_at) VALUES ($1,NOW())`, "SELECT id FROM users WHERE id=ANY%"},
  300. {"api_key_ingress", `UPDATE api_keys SET last_used_at=NOW() WHERE user_id=$1`, "SELECT user_id,key FROM api_keys WHERE user_id=ANY%"},
  301. } {
  302. t.Run(tc.name, func(t *testing.T) {
  303. db, svc, actor := cleanupTestDB(t)
  304. user := cleanupTestUser(t, db, "0")
  305. ctx := context.Background()
  306. preview, err := svc.Preview(ctx, actor)
  307. require.NoError(t, err)
  308. tx, err := db.Begin()
  309. require.NoError(t, err)
  310. defer func() { _ = tx.Rollback() }()
  311. _, err = tx.Exec(tc.mutation, user)
  312. require.NoError(t, err)
  313. type outcome struct {
  314. result *UserCleanupResult
  315. err error
  316. }
  317. done := make(chan outcome, 1)
  318. go func() {
  319. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  320. done <- outcome{result, err}
  321. }()
  322. require.Eventually(t, func() bool {
  323. var waiting bool
  324. err := db.QueryRow(`SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE $1)`, tc.waitingQuery).Scan(&waiting)
  325. return err == nil && waiting
  326. }, 10*time.Second, 50*time.Millisecond)
  327. require.NoError(t, tx.Commit())
  328. select {
  329. case result := <-done:
  330. require.NoError(t, result.err)
  331. require.Zero(t, result.result.DeletedCount)
  332. require.Equal(t, 1, result.result.SkippedCount)
  333. case <-time.After(20 * time.Second):
  334. t.Fatal("cleanup did not finish")
  335. }
  336. })
  337. }
  338. }
  339. func TestUserCleanupPostgresConcurrentDuplicateExecutions(t *testing.T) {
  340. db, svc, actor := cleanupTestDB(t)
  341. ctx := context.Background()
  342. cleanupTestUser(t, db, "0")
  343. preview, err := svc.Preview(ctx, actor)
  344. require.NoError(t, err)
  345. type outcome struct {
  346. result *UserCleanupResult
  347. err error
  348. }
  349. done := make(chan outcome, 2)
  350. start := make(chan struct{})
  351. for range 2 {
  352. go func() {
  353. <-start
  354. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  355. done <- outcome{result, err}
  356. }()
  357. }
  358. close(start)
  359. replays := 0
  360. for range 2 {
  361. select {
  362. case result := <-done:
  363. require.NoError(t, result.err)
  364. require.Equal(t, 1, result.result.DeletedCount)
  365. if result.result.Replayed {
  366. replays++
  367. }
  368. case <-time.After(20 * time.Second):
  369. t.Fatal("concurrent execution did not finish")
  370. }
  371. }
  372. require.Equal(t, 1, replays)
  373. var count int
  374. require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM audit_logs WHERE action='admin.users.cleanup'`).Scan(&count))
  375. require.Equal(t, 1, count)
  376. }
  377. func TestUserCleanupPostgresPreviewExpiryAndBoundedScope(t *testing.T) {
  378. db, svc, actor := cleanupTestDB(t)
  379. ctx := context.Background()
  380. cleanupTestUser(t, db, "0")
  381. preview, err := svc.Preview(ctx, actor)
  382. require.NoError(t, err)
  383. addedLater := cleanupTestUser(t, db, "0")
  384. _, err = db.Exec(`UPDATE user_cleanup_previews SET expires_at=NOW()-INTERVAL '1 second' WHERE id=$1`, preview.PreviewID)
  385. require.NoError(t, err)
  386. _, err = svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  387. require.ErrorIs(t, err, ErrUserCleanupPreview)
  388. _, err = db.Exec(`UPDATE user_cleanup_previews SET expires_at=NOW()+INTERVAL '1 minute' WHERE id=$1`, preview.PreviewID)
  389. require.NoError(t, err)
  390. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  391. require.NoError(t, err)
  392. require.Equal(t, 1, result.DeletedCount)
  393. var intact bool
  394. require.NoError(t, db.QueryRow(`SELECT deleted_at IS NULL FROM users WHERE id=$1`, addedLater).Scan(&intact))
  395. require.True(t, intact, "users outside the confirmed snapshot must remain intact")
  396. }
  397. func cleanupTestIdentityRelations(t *testing.T, db *sql.DB, userID int64) {
  398. t.Helper()
  399. _, err := db.Exec(`WITH added AS (INSERT INTO auth_identity_channels(identity_id,channel_subject) SELECT id,'channel-'||id::text FROM auth_identities WHERE user_id=$1 RETURNING identity_id) INSERT INTO identity_adoption_decisions(identity_id) SELECT identity_id FROM added`, userID)
  400. require.NoError(t, err)
  401. }
  402. func TestUserCleanupPostgresArchivesBindingsAndReleasesRegistrationIdentity(t *testing.T) {
  403. db, svc, actor := cleanupTestDB(t)
  404. ctx := context.Background()
  405. oldUser := cleanupTestUser(t, db, "0")
  406. cleanupTestIdentityRelations(t, db, oldUser)
  407. _, err := db.Exec(`UPDATE auth_identities SET metadata='{"profile_note":"private-identity-metadata"}'::jsonb WHERE user_id=$1`, oldUser)
  408. require.NoError(t, err)
  409. preview, err := svc.Preview(ctx, actor)
  410. require.NoError(t, err)
  411. result, err := svc.Execute(ctx, actor, preview.PreviewID, true, 1, UserCleanupAuditContext{})
  412. require.NoError(t, err)
  413. require.Equal(t, 1, result.DeletedCount)
  414. var archived bool
  415. require.NoError(t, db.QueryRow(`SELECT jsonb_array_length(identities)=1 AND jsonb_array_length(channels)=1 AND jsonb_array_length(adoption_decisions)=1 AND identities->0->'metadata'->>'profile_note'='private-identity-metadata' FROM user_cleanup_identity_archives WHERE user_id=$1`, oldUser).Scan(&archived))
  416. require.True(t, archived)
  417. var detached bool
  418. require.NoError(t, db.QueryRow(`SELECT COUNT(*)=1 AND COUNT(identity_id)=0 FROM identity_adoption_decisions`).Scan(&detached))
  419. require.True(t, detached, "adoption history retained with detached live binding")
  420. var newUser int64
  421. require.NoError(t, db.QueryRow(`INSERT INTO users(email) SELECT email FROM users WHERE id=$1 RETURNING id`, oldUser).Scan(&newUser))
  422. // The exact canonical identity uniqueness tuple used by repository registration
  423. // can be claimed by the newly registered same-email user after cleanup.
  424. _, err = db.Exec(`INSERT INTO auth_identities(user_id,provider_type,provider_key,provider_subject) SELECT id,'email','email',email FROM users WHERE id=$1`, newUser)
  425. require.NoError(t, err)
  426. var oldPreserved bool
  427. require.NoError(t, db.QueryRow(`SELECT deleted_at IS NOT NULL AND EXISTS(SELECT 1 FROM usage_logs WHERE user_id=$1) FROM users WHERE id=$1`, oldUser).Scan(&oldPreserved))
  428. require.True(t, oldPreserved)
  429. var leaked bool
  430. require.NoError(t, db.QueryRow(`SELECT EXISTS(SELECT 1 FROM audit_logs WHERE extra::text LIKE '%private-identity-metadata%') OR EXISTS(SELECT 1 FROM user_cleanup_previews WHERE candidate_snapshot::text LIKE '%private-identity-metadata%' OR result::text LIKE '%private-identity-metadata%')`).Scan(&leaked))
  431. require.False(t, leaked, "private identity archive must not leak into returned preview/result or operation logs")
  432. }