feat(effect-drizzle-sqlite): add vendored sqlite adapter (#28547)
This commit is contained in:
Kit Langton
@@ -0,0 +1,102 @@
|
||||
/* oxlint-disable */
|
||||
import * as Effect from "effect/Effect"
|
||||
import type { SqlError } from "effect/unstable/sql/SqlError"
|
||||
import { EffectDrizzleError } from "drizzle-orm/effect-core/errors"
|
||||
import type { QueryEffectHKTBase } from "drizzle-orm/effect-core/query-effect"
|
||||
import type { MigrationMeta } from "drizzle-orm/migrator"
|
||||
import { sql } from "drizzle-orm/sql/sql"
|
||||
import type { SQLiteEffectSession } from "../sqlite-core/effect/session"
|
||||
import {
|
||||
buildSQLiteMigrationBackfillStatements,
|
||||
prepareSQLiteMigrationBackfill,
|
||||
type SQLiteMigrationTableRow,
|
||||
} from "./sqlite"
|
||||
import { GET_VERSION_FOR, MIGRATIONS_TABLE_VERSIONS, type UpgradeResult } from "./utils"
|
||||
|
||||
const migrationUpgradeError = (cause: unknown) =>
|
||||
new EffectDrizzleError({
|
||||
message:
|
||||
typeof cause === "object" && cause !== null && "message" in cause && typeof cause.message === "string"
|
||||
? cause.message
|
||||
: String(cause),
|
||||
cause,
|
||||
})
|
||||
|
||||
export const upgradeIfNeeded: <TEffectHKT extends QueryEffectHKTBase>(
|
||||
migrationsTable: string,
|
||||
session: SQLiteEffectSession<TEffectHKT>,
|
||||
localMigrations: MigrationMeta[],
|
||||
) => Effect.Effect<UpgradeResult, EffectDrizzleError | TEffectHKT["error"] | SqlError, TEffectHKT["context"]> =
|
||||
Effect.fn("upgradeIfNeeded")(function* <TEffectHKT extends QueryEffectHKTBase>(
|
||||
migrationsTable: string,
|
||||
session: SQLiteEffectSession<TEffectHKT>,
|
||||
localMigrations: MigrationMeta[],
|
||||
) {
|
||||
const tableExists = yield* session.all(
|
||||
sql`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ${migrationsTable}`,
|
||||
)
|
||||
|
||||
if (tableExists.length === 0) {
|
||||
return { newDb: true }
|
||||
}
|
||||
|
||||
const rows = yield* session.all<{ column_name: string }>(
|
||||
sql`SELECT name as column_name FROM pragma_table_info(${migrationsTable})`,
|
||||
)
|
||||
|
||||
const version = GET_VERSION_FOR.sqlite(rows.map((r) => r.column_name))
|
||||
|
||||
for (let v = version; v < MIGRATIONS_TABLE_VERSIONS.sqlite; v++) {
|
||||
const upgradeFn = upgradeFunctions[v]
|
||||
if (!upgradeFn) {
|
||||
return yield* new EffectDrizzleError({
|
||||
message: `No upgrade path from migration table version ${v} to ${v + 1}`,
|
||||
cause: { version: v },
|
||||
})
|
||||
}
|
||||
yield* upgradeFn(migrationsTable, session, localMigrations)
|
||||
}
|
||||
|
||||
return { newDb: false }
|
||||
})
|
||||
|
||||
const upgradeFunctions: Record<
|
||||
number,
|
||||
<TEffectHKT extends QueryEffectHKTBase>(
|
||||
migrationsTable: string,
|
||||
session: SQLiteEffectSession<TEffectHKT>,
|
||||
localMigrations: MigrationMeta[],
|
||||
) => Effect.Effect<void, EffectDrizzleError | TEffectHKT["error"] | SqlError, TEffectHKT["context"]>
|
||||
> = {
|
||||
0: upgradeFromV0,
|
||||
}
|
||||
|
||||
function upgradeFromV0<TEffectHKT extends QueryEffectHKTBase>(
|
||||
migrationsTable: string,
|
||||
session: SQLiteEffectSession<TEffectHKT>,
|
||||
localMigrations: MigrationMeta[],
|
||||
): Effect.Effect<void, EffectDrizzleError | TEffectHKT["error"] | SqlError, TEffectHKT["context"]> {
|
||||
return Effect.gen(function* () {
|
||||
const table = sql`${sql.identifier(migrationsTable)}`
|
||||
|
||||
const dbRows = yield* session.all<SQLiteMigrationTableRow>(
|
||||
sql`SELECT id, hash, created_at FROM ${table} ORDER BY id ASC`,
|
||||
)
|
||||
const statements = yield* Effect.try({
|
||||
try: () =>
|
||||
buildSQLiteMigrationBackfillStatements(
|
||||
migrationsTable,
|
||||
prepareSQLiteMigrationBackfill(dbRows, localMigrations),
|
||||
),
|
||||
catch: migrationUpgradeError,
|
||||
})
|
||||
|
||||
yield* session.transaction((tx) =>
|
||||
Effect.gen(function* () {
|
||||
for (const statement of statements) {
|
||||
yield* tx.run(statement)
|
||||
}
|
||||
}),
|
||||
)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
/* oxlint-disable */
|
||||
import type { TablesRelationalConfig } from "drizzle-orm/_relations"
|
||||
import type { MigrationMeta } from "drizzle-orm/migrator"
|
||||
import type { AnyRelations } from "drizzle-orm/relations"
|
||||
import { type SQL, sql } from "drizzle-orm/sql/sql"
|
||||
import type { BaseSQLiteDatabase } from "drizzle-orm/sqlite-core"
|
||||
import type { SQLiteSession } from "drizzle-orm/sqlite-core/session"
|
||||
import { GET_VERSION_FOR, MIGRATIONS_TABLE_VERSIONS, type UpgradeResult } from "./utils"
|
||||
|
||||
/** @internal */
|
||||
export type SQLiteMigrationTableRow = { id: number | null; hash: string; created_at: number }
|
||||
|
||||
type AsyncSQLiteDatabaseWithSession = BaseSQLiteDatabase<"async", unknown, Record<string, unknown>> & {
|
||||
session: {
|
||||
all<T>(query: SQL): Promise<T[]>
|
||||
}
|
||||
transaction<T>(transaction: (tx: { run(query: SQL): Promise<unknown> }) => Promise<T>): Promise<T>
|
||||
}
|
||||
|
||||
type SQLiteMigrationBackfillEntry = {
|
||||
name: string
|
||||
selector:
|
||||
| { column: "id"; value: number }
|
||||
| { column: "created_at"; value: number }
|
||||
| { column: "hash"; value: string }
|
||||
}
|
||||
|
||||
function unmatchedMigrationError(unmatched: SQLiteMigrationTableRow[]) {
|
||||
return new Error(
|
||||
`While upgrading your database migrations table we found ${unmatched.length} (${unmatched
|
||||
.map((it) => `[id: ${it.id}, created_at: ${it.created_at}]`)
|
||||
.join(
|
||||
", ",
|
||||
)}) migrations in the database that do not match any local migration. This means that some migrations were applied to the database but are missing from the local environment`,
|
||||
)
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export function prepareSQLiteMigrationBackfill(
|
||||
dbRows: SQLiteMigrationTableRow[],
|
||||
localMigrations: MigrationMeta[],
|
||||
): SQLiteMigrationBackfillEntry[] {
|
||||
const sortedLocalMigrations = [...localMigrations].sort((a, b) =>
|
||||
a.folderMillis !== b.folderMillis ? a.folderMillis - b.folderMillis : (a.name ?? "").localeCompare(b.name ?? ""),
|
||||
)
|
||||
const byMillis = new Map<number, MigrationMeta[]>()
|
||||
const byHash = new Map<string, MigrationMeta>()
|
||||
for (const migration of sortedLocalMigrations) {
|
||||
if (!byMillis.has(migration.folderMillis)) {
|
||||
byMillis.set(migration.folderMillis, [])
|
||||
}
|
||||
byMillis.get(migration.folderMillis)!.push(migration)
|
||||
byHash.set(migration.hash, migration)
|
||||
}
|
||||
|
||||
const toApply: SQLiteMigrationBackfillEntry[] = []
|
||||
const unmatched: SQLiteMigrationTableRow[] = []
|
||||
|
||||
for (const dbRow of dbRows) {
|
||||
const stringified = String(dbRow.created_at)
|
||||
const millis = Number(stringified.substring(0, stringified.length - 3) + "000")
|
||||
const candidates = byMillis.get(millis)
|
||||
|
||||
const matchedByMillis = candidates?.length === 1 ? candidates[0] : undefined
|
||||
const matchedByCandidateHash =
|
||||
candidates && candidates.length > 1
|
||||
? candidates.find((candidate) => candidate.hash && dbRow.hash && candidate.hash === dbRow.hash)
|
||||
: undefined
|
||||
const matchedByHash = matchedByMillis || matchedByCandidateHash ? undefined : byHash.get(dbRow.hash)
|
||||
const matched = matchedByMillis ?? matchedByCandidateHash ?? matchedByHash
|
||||
|
||||
if (matched) {
|
||||
toApply.push({
|
||||
name: matched.name,
|
||||
selector:
|
||||
dbRow.id !== null
|
||||
? { column: "id", value: dbRow.id }
|
||||
: matchedByMillis
|
||||
? { column: "created_at", value: dbRow.created_at }
|
||||
: { column: "hash", value: dbRow.hash },
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
unmatched.push(dbRow)
|
||||
}
|
||||
|
||||
if (unmatched.length > 0) {
|
||||
throw unmatchedMigrationError(unmatched)
|
||||
}
|
||||
|
||||
return toApply
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export function buildSQLiteMigrationBackfillStatements(
|
||||
migrationsTable: string,
|
||||
backfillEntries: SQLiteMigrationBackfillEntry[],
|
||||
) {
|
||||
const table = sql`${sql.identifier(migrationsTable)}`
|
||||
const statements: SQL[] = [
|
||||
sql`ALTER TABLE ${table} ADD COLUMN ${sql.identifier("name")} text`,
|
||||
sql`ALTER TABLE ${table} ADD COLUMN ${sql.identifier("applied_at")} TEXT`,
|
||||
]
|
||||
|
||||
for (const backfillEntry of backfillEntries) {
|
||||
const updateQuery = sql`UPDATE ${table} SET ${sql.identifier("name")} = ${backfillEntry.name}, ${sql.identifier(
|
||||
"applied_at",
|
||||
)} = NULL WHERE`
|
||||
|
||||
updateQuery.append(sql` ${sql.identifier(backfillEntry.selector.column)} = ${backfillEntry.selector.value}`)
|
||||
|
||||
statements.push(updateQuery)
|
||||
}
|
||||
|
||||
return statements
|
||||
}
|
||||
|
||||
/**
|
||||
* Detects the current version of the migrations table schema and upgrades it if needed.
|
||||
*
|
||||
* Version 0: Original schema (id, hash, created_at)
|
||||
* Version 1: Extended schema (id, hash, created_at, name, applied_at)
|
||||
*/
|
||||
export function upgradeSyncIfNeeded(
|
||||
migrationsTable: string,
|
||||
session: SQLiteSession<"sync", unknown, Record<string, unknown>, AnyRelations, TablesRelationalConfig>,
|
||||
localMigrations: MigrationMeta[],
|
||||
): UpgradeResult {
|
||||
const tableExists = session.all(sql`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ${migrationsTable}`)
|
||||
|
||||
if (tableExists.length === 0) {
|
||||
return { newDb: true }
|
||||
}
|
||||
|
||||
// Table exists, check table shape
|
||||
const rows = session.all<{ column_name: string }>(
|
||||
sql`SELECT name as column_name FROM pragma_table_info(${migrationsTable})`,
|
||||
)
|
||||
|
||||
const version = GET_VERSION_FOR.sqlite(rows.map((r) => r.column_name))
|
||||
|
||||
for (let v = version; v < MIGRATIONS_TABLE_VERSIONS.sqlite; v++) {
|
||||
const upgradeFn = upgradeSyncFunctions[v]
|
||||
if (!upgradeFn) {
|
||||
throw new Error(`No upgrade path from migration table version ${v} to ${v + 1}`)
|
||||
}
|
||||
upgradeFn(migrationsTable, session, localMigrations)
|
||||
}
|
||||
|
||||
return { newDb: false }
|
||||
}
|
||||
|
||||
const upgradeSyncFunctions: Record<
|
||||
number,
|
||||
(
|
||||
migrationsTable: string,
|
||||
session: SQLiteSession<"sync", unknown, Record<string, unknown>, AnyRelations, TablesRelationalConfig>,
|
||||
localMigrations: MigrationMeta[],
|
||||
) => void
|
||||
> = {
|
||||
/**
|
||||
* Upgrade from version 0 to version 1:
|
||||
* 1. Read all existing DB migrations
|
||||
* 2. Sort localMigrations ASC by millis and if the same - sort by name
|
||||
* 3. Match each DB row to a local migration
|
||||
* If multiple migrations share the same second, use hash matching as a tiebreaker
|
||||
* Not implemented for now -> If hash matching fails, fall back to serial id ordering
|
||||
* 5. Create extra column and backfill names for matched migrations
|
||||
*/
|
||||
0: (migrationsTable, session, localMigrations) => {
|
||||
const table = sql`${sql.identifier(migrationsTable)}`
|
||||
const dbRows = session.all<SQLiteMigrationTableRow>(sql`SELECT id, hash, created_at FROM ${table} ORDER BY id ASC`)
|
||||
const statements = buildSQLiteMigrationBackfillStatements(
|
||||
migrationsTable,
|
||||
prepareSQLiteMigrationBackfill(dbRows, localMigrations),
|
||||
)
|
||||
|
||||
session.transaction((tx) => {
|
||||
for (const statement of statements) {
|
||||
tx.run(statement)
|
||||
}
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
/**
|
||||
* Detects the current version of the migrations table schema and upgrades it if needed.
|
||||
*
|
||||
* Version 0: Original schema (id, hash, created_at)
|
||||
* Version 1: Extended schema (id, hash, created_at, name, applied_at)
|
||||
*/
|
||||
export async function upgradeAsyncIfNeeded(
|
||||
migrationsTable: string,
|
||||
db: AsyncSQLiteDatabaseWithSession,
|
||||
localMigrations: MigrationMeta[],
|
||||
): Promise<UpgradeResult> {
|
||||
// Check if the table exists at all
|
||||
const tableExists = await db.session.all(
|
||||
sql`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ${migrationsTable}`,
|
||||
)
|
||||
|
||||
if (tableExists.length === 0) {
|
||||
return { newDb: true }
|
||||
}
|
||||
|
||||
const rows = await db.session.all<{ column_name: string }>(
|
||||
sql`SELECT name as column_name FROM pragma_table_info(${migrationsTable})`,
|
||||
)
|
||||
|
||||
const version = GET_VERSION_FOR.sqlite(rows.map((r) => r.column_name))
|
||||
|
||||
for (let v = version; v < MIGRATIONS_TABLE_VERSIONS.sqlite; v++) {
|
||||
const upgradeFn = upgradeAsyncFunctions[v]
|
||||
if (!upgradeFn) {
|
||||
throw new Error(`No upgrade path from migration table version ${v} to ${v + 1}`)
|
||||
}
|
||||
await upgradeFn(migrationsTable, db, localMigrations)
|
||||
}
|
||||
|
||||
return { newDb: false }
|
||||
}
|
||||
|
||||
const upgradeAsyncFunctions: Record<
|
||||
number,
|
||||
(migrationsTable: string, db: AsyncSQLiteDatabaseWithSession, localMigrations: MigrationMeta[]) => Promise<void>
|
||||
> = {
|
||||
/**
|
||||
* Upgrade from version 0 to version 1:
|
||||
* 1. Read all existing DB migrations
|
||||
* 2. Sort localMigrations ASC by millis and if the same - sort by name
|
||||
* 3. Match each DB row to a local migration
|
||||
* If multiple migrations share the same second, use hash matching as a tiebreaker
|
||||
* Not implemented for now -> If hash matching fails, fall back to serial id ordering
|
||||
* 5. Create extra column and backfill names for matched migrations
|
||||
*/
|
||||
0: async (migrationsTable, db, localMigrations) => {
|
||||
const table = sql`${sql.identifier(migrationsTable)}`
|
||||
const dbRows = await db.session.all<SQLiteMigrationTableRow>(
|
||||
sql`SELECT id, hash, created_at FROM ${table} ORDER BY id ASC`,
|
||||
)
|
||||
const statements = buildSQLiteMigrationBackfillStatements(
|
||||
migrationsTable,
|
||||
prepareSQLiteMigrationBackfill(dbRows, localMigrations),
|
||||
)
|
||||
|
||||
await db.transaction(async (tx) => {
|
||||
for (const statement of statements) {
|
||||
await tx.run(statement)
|
||||
}
|
||||
})
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
/* oxlint-disable */
|
||||
export interface UpgradeResult {
|
||||
newDb: boolean
|
||||
}
|
||||
|
||||
export const MIGRATIONS_TABLE_VERSIONS = {
|
||||
sqlite: 1,
|
||||
pg: 1,
|
||||
effect: 1,
|
||||
mysql: 1,
|
||||
mssql: 1,
|
||||
cockroach: 1,
|
||||
singlestore: 1,
|
||||
} as const
|
||||
|
||||
export const GET_VERSION_FOR = {
|
||||
mysql: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
pg: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
effect: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
mssql: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
cockroach: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
singlestore: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
sqlite: (columns: string[]): number => {
|
||||
if (columns.includes("name")) return 1
|
||||
return 0
|
||||
},
|
||||
} as const
|
||||
Reference in New Issue
Block a user