107 lines
3.7 KiB
TypeScript
107 lines
3.7 KiB
TypeScript
import { assert, it } from "@effect/vitest"
|
|
import { Effect, Exit, Fiber, Latch, Layer, Schema } from "effect"
|
|
import * as PersistedCacheTest from "effect-test/unstable/persistence/PersistedCacheTest"
|
|
import * as PersistedQueueTest from "effect-test/unstable/persistence/PersistedQueueTest"
|
|
import { TestClock } from "effect/testing"
|
|
import { PersistedQueue, Persistence } from "effect/unstable/persistence"
|
|
import { SqlClient } from "effect/unstable/sql"
|
|
import { PgContainer } from "./utils.ts"
|
|
|
|
PersistedCacheTest.suite(
|
|
"sql-pg-multi",
|
|
Persistence.layerSqlMultiTable.pipe(Layer.provide(PgContainer.layerClient))
|
|
)
|
|
|
|
PersistedCacheTest.suite(
|
|
"sql-pg-single",
|
|
Persistence.layerSql.pipe(Layer.provide(PgContainer.layerClient))
|
|
)
|
|
|
|
PersistedQueueTest.suite(
|
|
"sql-pg",
|
|
PersistedQueue.layerStoreSql().pipe(Layer.provide(PgContainer.layerClient))
|
|
)
|
|
|
|
it.layer(PgContainer.layerClient, { timeout: "30 seconds" })("PersistedQueue SQL locks", (it) => {
|
|
it.effect("refreshes locks for acquired elements", () =>
|
|
Effect.gen(function*() {
|
|
const options = {
|
|
tableName: "effect_queue_lock_refresh",
|
|
pollInterval: "10 millis",
|
|
lockRefreshInterval: "100 millis",
|
|
lockExpiration: "1 second"
|
|
} as const
|
|
const store1 = yield* PersistedQueue.makeStoreSql(options)
|
|
const store2 = yield* PersistedQueue.makeStoreSql(options)
|
|
const element = { message: "hello" }
|
|
|
|
yield* store1.offer({
|
|
name: "lock-refresh",
|
|
id: crypto.randomUUID(),
|
|
element,
|
|
isCustomId: false
|
|
})
|
|
|
|
const acquired = Latch.makeUnsafe()
|
|
const first = yield* Effect.scoped(Effect.gen(function*() {
|
|
yield* store1.take({ name: "lock-refresh", maxAttempts: 10 })
|
|
yield* acquired.open
|
|
return yield* Effect.never
|
|
})).pipe(Effect.forkScoped)
|
|
|
|
yield* acquired.await
|
|
|
|
const second = yield* Effect.scoped(
|
|
store2.take({ name: "lock-refresh", maxAttempts: 10 })
|
|
).pipe(Effect.forkScoped)
|
|
|
|
yield* Effect.sleep("1500 millis")
|
|
assert.isUndefined(second.pollUnsafe())
|
|
|
|
yield* Fiber.interrupt(first)
|
|
const received = yield* Fiber.join(second)
|
|
assert.deepStrictEqual(received.element, element)
|
|
}).pipe(TestClock.withLive))
|
|
|
|
it.effect("counts malformed JSON as an attempt and continues", () =>
|
|
Effect.gen(function*() {
|
|
const tableName = "effect_queue_invalid_json"
|
|
const store = yield* PersistedQueue.makeStoreSql({
|
|
tableName,
|
|
pollInterval: "10 millis"
|
|
})
|
|
const factory = yield* PersistedQueue.makeFactory.pipe(
|
|
Effect.provideService(PersistedQueue.PersistedQueueStore, store)
|
|
)
|
|
const queue = yield* factory.make({
|
|
name: "invalid-json",
|
|
schema: Schema.String
|
|
})
|
|
const sql = (yield* SqlClient.SqlClient).withoutTransforms()
|
|
const table = sql(tableName)
|
|
const poisonId = crypto.randomUUID()
|
|
|
|
yield* store.offer({
|
|
name: "invalid-json",
|
|
id: poisonId,
|
|
element: "poison",
|
|
isCustomId: false
|
|
})
|
|
yield* sql`UPDATE ${table} SET element = ${"{"} WHERE id = ${poisonId}`
|
|
yield* queue.offer("valid")
|
|
|
|
const malformed = yield* Effect.exit(queue.take(Effect.succeed, { maxAttempts: 1 }))
|
|
assert.isTrue(Exit.isFailure(malformed))
|
|
|
|
const rows = yield* sql<{
|
|
readonly attempts: number
|
|
readonly last_failure: string | null
|
|
}>`SELECT attempts, last_failure FROM ${table} WHERE id = ${poisonId}`
|
|
assert.strictEqual(rows[0].attempts, 1)
|
|
assert.isNotNull(rows[0].last_failure)
|
|
|
|
const value = yield* queue.take(Effect.succeed, { maxAttempts: 1 })
|
|
assert.strictEqual(value, "valid")
|
|
}).pipe(TestClock.withLive))
|
|
})
|