1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Layer, Schema } from "effect"
3import * as PersistedCacheTest from "effect-test/persistence/PersistedCacheTest"
4import * as PersistedQueueTest from "effect-test/persistence/PersistedQueueTest"
5import * as SqlCleanupTest from "effect-test/persistence/SqlCleanupTest"
6import { PersistedQueue, Persistence } from "effect/persistence"
7import { SqlClient } from "effect/sql"
8import { TestClock } from "effect/testing"
9import { MysqlContainer } from "./utils.ts"
10
11it.layer(MysqlContainer.layerClient, { timeout: "90 seconds" })("Persistence", (it) => {
12 PersistedCacheTest.suiteWith("sql-mysql2-multi", Persistence.layerSqlMultiTable, it)
13
14 PersistedQueueTest.suiteWith("sql-mysql2", PersistedQueue.layerStoreSql(), it)
15
16 // elements are stored in a MEDIUMTEXT column, so payloads must survive the
17 // 64KB TEXT limit
18 it.effect("round-trips queue payloads larger than 64KB", () =>
19 Effect.gen(function*() {
20 const store = yield* PersistedQueue.makeStoreSql({
21 tableName: "effect_queue_large_payload",
22 pollInterval: "10 millis"
23 })
24 const factory = yield* PersistedQueue.makeFactory.pipe(
25 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
26 )
27 const queue = yield* factory.make({ name: "large-payload", schema: Schema.String })
28
29 const payload = "x".repeat(200_000)
30 yield* queue.offer(payload)
31 const value = yield* queue.take(Effect.succeed)
32 assert.strictEqual(value, payload)
33 }).pipe(TestClock.withLive), { timeout: 30000 })
34
35 describe("single-table persistence", { concurrent: false }, () => {
36 PersistedCacheTest.suiteWith("sql-mysql2-single", Persistence.layerSql, it)
37
38 it.effect("deletes expired entries in batches", () =>
39 Effect.gen(function*() {
40 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
41 const table = sql("effect_persistence")
42 const expiredCount = sql<{ readonly count: number }>`
43 SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'expired'
44 `.pipe(Effect.map((rows) => Number(rows[0].count)))
45 // Reset the table left by the single-table cache suite so cleanup builds its own schema and index.
46 yield* sql`DROP TABLE IF EXISTS ${table}`
47 yield* sql`
48 CREATE TABLE ${table} (
49 store_id VARCHAR(191) NOT NULL,
50 id VARCHAR(191) NOT NULL,
51 value TEXT NOT NULL,
52 expires BIGINT,
53 PRIMARY KEY (store_id, id)
54 )
55 `
56
57 const entries = Array.from({ length: SqlCleanupTest.expiredEntryCount }, (_, i) => ({
58 store_id: "expired",
59 id: String(i),
60 value: "{}",
61 expires: SqlCleanupTest.expiredAtEpoch
62 }))
63 yield* sql`INSERT INTO ${table} ${sql.insert(entries)}`.unprepared
64 yield* sql`
65 INSERT INTO ${table} (store_id, id, value, expires)
66 VALUES ('live', 'live', '{}', NULL), ('live', 'future', '{}', ${SqlCleanupTest.futureExpiresAt})
67 `
68
69 yield* Layer.build(Persistence.layerBackingSql).pipe(TestClock.withLive)
70
71 const expired = yield* SqlCleanupTest.waitForCount(expiredCount, (count) => count === 0)
72 assert.strictEqual(expired, 0)
73 const live = yield* sql<{ readonly count: number }>`
74 SELECT COUNT(*) AS count FROM ${table} WHERE store_id = 'live'
75 `
76 assert.strictEqual(Number(live[0].count), 2)
77
78 const indexes = yield* sql<{ readonly count: number }>`
79 SELECT COUNT(*) AS count FROM information_schema.statistics
80 WHERE table_schema = DATABASE()
81 AND table_name = 'effect_persistence'
82 AND index_name = 'effect_persistence_expires_idx'
83 `
84 assert.strictEqual(Number(indexes[0].count), 1)
85 }), { timeout: SqlCleanupTest.testTimeout })
86 })
87})