Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fluffy-pillows-report.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@effect/sql": minor
---

feat: add support for Optimistic Concurrency via `versionColumn` option.
58 changes: 41 additions & 17 deletions packages/sql/src/Model.ts
Original file line number Diff line number Diff line change
Expand Up @@ -688,6 +688,7 @@ export const makeRepository = <
readonly tableName: string
readonly spanPrefix: string
readonly idColumn: Id
readonly versionColumn?: string | undefined
}): Effect.Effect<
{
readonly insert: (
Expand Down Expand Up @@ -716,6 +717,7 @@ export const makeRepository = <
const sql = yield* SqlClient
const idSchema = Model.fields[options.idColumn] as Schema.Schema.Any
const idColumn = options.idColumn as string
const versionColumn = options.versionColumn

const insertSchema = SqlSchema.single({
Request: Model.insert,
Expand Down Expand Up @@ -759,20 +761,37 @@ select * from ${sql(options.tableName)} where ${sql(idColumn)} = LAST_INSERT_ID(
const updateSchema = SqlSchema.single({
Request: Model.update,
Result: Model,
execute: (request) =>
sql.onDialectOrElse({
mysql: () =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
};
execute: versionColumn ?
(request) =>
sql.onDialectOrElse({
mysql: () =>
sql`update ${sql(options.tableName)} set ${
sql.update({ ...request, [versionColumn]: Uuid.v4() }, [idColumn])
} where ${sql(idColumn)} = ${request[idColumn]} and ${sql(versionColumn)} = ${request[versionColumn]};
select * from ${sql(options.tableName)} where ${sql(idColumn)} = ${request[idColumn]};`.unprepared.pipe(
Effect.map(([, results]) => results as any)
),
orElse: () =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
} returning *`
})
Effect.map(([, results]) => results as any)
),
orElse: () =>
sql`update ${sql(options.tableName)} set ${
sql.update({ ...request, [versionColumn]: Uuid.v4() }, [idColumn])
} where ${sql(idColumn)} = ${request[idColumn]} and ${sql(versionColumn)} = ${
request[versionColumn]
} returning *`
}) :
(request) =>
sql.onDialectOrElse({
mysql: () =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
};
select * from ${sql(options.tableName)} where ${sql(idColumn)} = ${request[idColumn]};`.unprepared.pipe(
Effect.map(([, results]) => results as any)
),
orElse: () =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
} returning *`
})
})
const update = (
update: S["update"]["Type"]
Expand All @@ -787,10 +806,15 @@ select * from ${sql(options.tableName)} where ${sql(idColumn)} = ${request[idCol

const updateVoidSchema = SqlSchema.void({
Request: Model.update,
execute: (request) =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
}`
execute: versionColumn ?
(request) =>
sql`update ${sql(options.tableName)} set ${
sql.update({ ...request, [versionColumn]: Uuid.v4() }, [idColumn])
} where ${sql(idColumn)} = ${request[idColumn]} and ${sql(versionColumn)} = ${request[versionColumn]}`
: (request) =>
sql`update ${sql(options.tableName)} set ${sql.update(request, [idColumn])} where ${sql(idColumn)} = ${
request[idColumn]
}`
})
const updateVoid = (
update: S["update"]["Type"]
Expand Down
Loading