Skip to content
Merged
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
42 changes: 42 additions & 0 deletions .changeset/21113-turso-remote-autonumber.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
---
'@objectstack/driver-turso': minor
'@objectstack/driver-sql': patch
---

feat(driver-turso): the remote transport issues `auto_number` values (#21113)

Clause-②: yes (widening)

A `create()`, `bulkCreate()` or `upsert()` on the Turso REMOTE transport that
leaves an `autonumber` field empty (`undefined`, `null` or `''`) now gets a
generated value. It used to be refused with `NOT_IMPLEMENTED` / 501, so on a
hosted tenant database — which is on this transport — no object declaring an
`auto_number` field could get a new record at all. Nothing is declared anew in
the spec or in the package exports; `supports.autonumber` stays `true` and is
now honoured.

- The value comes from the same persistent `_objectstack_sequences` counter the
local and embedded-replica transports use, rendered by the same format rules
(`autonumberFormat` / `format`, organization scope, date and `{field}`
tokens), bootstrapped from the table's highest existing value by the same
reading, and re-seeded the same way after rows land above the counter by a
seed replay or import. A remote driver and an embedded replica of one
database draw from one counter row.
- The counter moves in one statement over the connection (`UPDATE … RETURNING`,
or on a cold counter `INSERT … ON CONFLICT (key_hash) DO UPDATE … RETURNING`),
so writers in different processes never draw the same number.
- An `upsert()` that merges into an existing row keeps the number already in
the row; a row that carries its own number is written unchanged.
- A `_objectstack_sequences` table in the pre-`key_hash` shape is refused in
remote mode with `DATABASE_ERROR` / 500 and the remedy in the message (open
the database once through the local or embedded-replica transport, which
migrates it); remote mode does not migrate it and does not key by the legacy
rule.
- `RemoteTransport.upsert()` takes an optional fifth argument naming columns
that are written on insert and left alone on merge.

`@objectstack/driver-sql`: the sequence rules a second transport shares are
now `protected` members of `SqlDriver` (`resolveSequenceTenantId`,
`defineSequencesTable`, `maxAutonumberCounter`, `escapeLikePrefix`,
`sequencesTableName`, `autoNumberCollisionRetries`). No export is added and no
behaviour changes on any dialect.
96 changes: 76 additions & 20 deletions packages/drivers/driver-sql/src/sql-driver.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7347,6 +7347,56 @@ export class SqlDriver implements IDataDriver {
.digest('hex');
}

// ── The sequence semantics a second face shares, stated once ──────────────
//
// `TursoDriver`'s REMOTE face issues record numbers over `@libsql/client`
// rather than through this class's Knex transaction, and an embedded-replica
// face and a remote face can point at ONE database. Two faces computing a
// different counter key, a different tenant bucket or a different data-table
// bootstrap would hand out colliding numbers on that database, so each rule
// below is a `protected` member that face CALLS — never a copy it keeps. The
// members are small on purpose: they are the whole of what decides which
// counter a value is drawn from and where a cold counter starts.

/** The sequence-counter table's name, for a face that spells its own statements against it. */
protected get sequencesTableName(): string {
return SEQUENCES_TABLE;
}

/** How many times a write re-seeds and retries on a provable autonumber collision (#5495). */
protected get autoNumberCollisionRetries(): number {
return AUTONUMBER_COLLISION_RETRIES;
}

/**
* The tenant bucket a counter row is keyed by: the row's tenant when the
* object HAS a tenant column and the row carries one, else the platform
* bucket `GLOBAL_TENANT`. One spelling for the issue path, the re-seed path
* and the counter identity, which used to carry it three times.
*/
protected resolveSequenceTenantId(tenantField: string | null, tenantId: string | null): string {
return tenantField && tenantId ? String(tenantId) : GLOBAL_TENANT;
}

/**
* The column definitions of the `key_hash`-keyed sequences table — the ONE
* place the shape is spelled. {@link createSequencesTable} runs it through a
* live connection; a face with no Knex connection compiles the same builder
* to text (`knex.schema.createTable(name, t => this.defineSequencesTable(t)).toSQL()`)
* and sends that, so both faces create byte-for-byte the same table.
*/
protected defineSequencesTable(t: Knex.CreateTableBuilder): void {
t.string('key_hash', 64).notNullable().primary();
t.string('object').notNullable();
t.string('tenant_id').notNullable();
t.string('field').notNullable();
// Non-indexed, so it is free of the PK length limit — a long `{plan_no}`
// composite scope fits. 1024 is far above any realistic rendered prefix.
t.string('scope', 1024).notNullable().defaultTo('');
t.bigInteger('last_value').notNullable().defaultTo(0);
t.timestamp('updated_at').defaultTo(this.knex.fn.now());
}

/**
* Create the current `key_hash`-keyed sequences table shape. `runner` is the
* connection the DDL runs on (a fresh pooled connection by default, or the
Expand All @@ -7356,17 +7406,7 @@ export class SqlDriver implements IDataDriver {
table: string,
runner: Knex | Knex.Transaction = this.knex,
): Promise<void> {
await runner.schema.createTable(table, (t) => {
t.string('key_hash', 64).notNullable().primary();
t.string('object').notNullable();
t.string('tenant_id').notNullable();
t.string('field').notNullable();
// Non-indexed, so it is free of the PK length limit — a long `{plan_no}`
// composite scope fits. 1024 is far above any realistic rendered prefix.
t.string('scope', 1024).notNullable().defaultTo('');
t.bigInteger('last_value').notNullable().defaultTo(0);
t.timestamp('updated_at').defaultTo(this.knex.fn.now());
});
await runner.schema.createTable(table, (t) => this.defineSequencesTable(t));
}

/**
Expand Down Expand Up @@ -7497,16 +7537,34 @@ export class SqlDriver implements IDataDriver {
tenantId: string | null,
suffix = '',
): Promise<number> {
const escapedPrefix = prefix.replace(/([\\%_])/g, '\\$1');
let builder = queryRunner(tableName).select(field).where(field, 'like', `${escapedPrefix}%`).whereNotNull(field);
let builder = queryRunner(tableName).select(field).where(field, 'like', `${this.escapeLikePrefix(prefix)}%`).whereNotNull(field);
if (tenantField && tenantId !== null) {
builder = builder.where(tenantField, tenantId);
}
const rows = await builder;
return this.maxAutonumberCounter((rows as any[]).map((r) => (r as any)[field]), prefix, suffix);
}

/**
* The rendered prefix as a `LIKE` anchor: `\`, `%` and `_` escaped with a
* backslash, so a prefix is matched literally. The predicate's pre-filter
* only — {@link maxAutonumberCounter} re-checks the prefix per row.
*/
protected escapeLikePrefix(prefix: string): string {
return prefix.replace(/([\\%_])/g, '\\$1');
}

/**
* The bootstrap reading itself — the highest counter among the stored values
* of one counter's partition, under the anchored / unanchored rules the
* {@link scanMaxNumericTail} docblock states. Separated from the statement
* that fetches the values so a face that fetches them through another
* transport reads them by this one rule rather than by a copy of it.
*/
protected maxAutonumberCounter(values: Iterable<unknown>, prefix: string, suffix: string): number {
let maxN = 0;
const anchored = prefix !== '' || suffix !== '';
for (const r of rows as any[]) {
const v: string = (r as any)[field];
for (const v of values) {
if (typeof v !== 'string') continue;
let n: number;
if (anchored) {
Expand Down Expand Up @@ -7647,7 +7705,7 @@ export class SqlDriver implements IDataDriver {
// locking on a second one (SQLite pool max=1). `initObjects` normally warms
// this up front, making the call a no-op — this only bites the lazy path.
await this.ensureSequencesTable(parentTrx);
const resolvedTenantId = tenantField && tenantId ? String(tenantId) : GLOBAL_TENANT;
const resolvedTenantId = this.resolveSequenceTenantId(tenantField, tenantId);
if (scope !== '' && !this.sequencesHasKeyHash) {
// The legacy sequences table could not be migrated to the key_hash shape,
// so it cannot represent per-scope counters. Fail with a clear, actionable
Expand Down Expand Up @@ -7907,8 +7965,7 @@ export class SqlDriver implements IDataDriver {
*/
protected async resyncSequenceToDataMax(reservation: AutoNumberReservation): Promise<void> {
await this.ensureSequencesTable();
const resolvedTenantId =
reservation.tenantField && reservation.tenantId ? String(reservation.tenantId) : GLOBAL_TENANT;
const resolvedTenantId = this.resolveSequenceTenantId(reservation.tenantField, reservation.tenantId);
const key = this.sequencesHasKeyHash
? { key_hash: this.sequenceKeyHash(reservation.tableName, resolvedTenantId, reservation.field, reservation.scope) }
: { object: reservation.tableName, tenant_id: resolvedTenantId, field: reservation.field };
Expand Down Expand Up @@ -7942,8 +7999,7 @@ export class SqlDriver implements IDataDriver {
* that drew from it. See {@link bulkCreate}.
*/
protected autoNumberCounterKey(reservation: AutoNumberReservation): string {
const resolvedTenantId =
reservation.tenantField && reservation.tenantId ? String(reservation.tenantId) : GLOBAL_TENANT;
const resolvedTenantId = this.resolveSequenceTenantId(reservation.tenantField, reservation.tenantId);
return this.sequenceKeyHash(reservation.tableName, resolvedTenantId, reservation.field, reservation.scope);
}

Expand Down
25 changes: 18 additions & 7 deletions packages/drivers/driver-turso/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,6 @@ table lists the ones that answer differently from local mode.
| Operation | Refused call | Use instead |
|:---|:---|:---|
| Transactions | `beginTransaction()`, `commit()`, `rollback()` (and their deprecated aliases `commitTransaction()` and `rollbackTransaction()`), and `options.transaction` passed to any `RemoteTransport` method in the tree above, to `aggregate()` or to `syncSchemasBatch()` | The local or embedded-replica transport, which run Knex transactions and honour `options.transaction` |
| Record numbers | `create()`, `bulkCreate()`, and an `upsert()` with no `id`, `_id` or `conflictKeys`, when a row leaves an `autonumber` field empty (`undefined`, `null` or `''`) | The local or embedded-replica transport, which generate record numbers, or a value you supply, which is written unchanged |
| Deferring schema DDL | `setDeferredDdl(true)`, which `os migrate plan` calls. `setDeferredDdl(false)` is accepted | Run the command against a local SQLite copy of the database (a `file:` URL) |
| Schema drift detection | `detectManagedDrift()` | `os migrate plan` against a local SQLite copy of the database (a `file:` URL) |
| Planning the ADR-0104 media column move | `planMediaColumnMove()`, the column step of `os migrate files-to-references` | The local or embedded-replica transport, which plan it |
Expand All @@ -147,18 +146,30 @@ table lists the ones that answer differently from local mode.
reads that and does not call `beginTransaction()`: it runs the callback with
no transaction and logs a warning once per datasource, or, called with
`require: true`, throws `TransactionUnsupportedError` before the callback runs.
- **Record numbers.** Remote mode never generates one. The check uses the
`autonumber` fields this driver recorded when it synced the object's schema
(`syncSchema`, `syncSchemasBatch` or `initObjects`). An `upsert()` that
carries an `id`, `_id` or `conflictKeys` is not refused, because it may merge
into an existing row. If it inserts instead, the row is written with the field
empty and the driver logs a warning naming the row.
- **Aggregation.** `aggregate()` called on the driver directly also refuses,
with the same code, a `groupBy` entry that has a `dateGranularity`, and an
`aggregations` entry with a non-empty `filter`. `engine.aggregate()` never
sends either one to this driver: it fetches the rows and computes both in
memory. The local transport also refuses a per-aggregation `filter`.

#### Record numbers in remote mode

An `autonumber` field a row leaves empty (`undefined`, `null` or `''`) is
filled in remote mode as it is in local and embedded-replica mode: `create()`,
`bulkCreate()` and `upsert()` issue the next value from the persistent
`_objectstack_sequences` counter, rendered with the field's `autonumberFormat`
(or `format`) and scoped per organization, per date token and per `{field}`
token exactly as `SqlDriver` scopes it. The counter moves in one statement over
the connection (`UPDATE … RETURNING`; on a cold counter, a bootstrap from the
table's highest existing value followed by `INSERT … ON CONFLICT DO UPDATE …
RETURNING`), so two processes writing to one database never draw the same
number. A row that already carries a value keeps it (a seed replay or an
import); an `upsert()` that merges into an existing row keeps the number
already in the row. A `_objectstack_sequences` table in the legacy shape
(created by a local-mode driver older than the `key_hash` column) is refused
with `code: 'DATABASE_ERROR'` / `status: 500` and the remedy in the message;
remote mode does not migrate it.

Answered in remote mode, differently from local mode:

- **`distinct()`** runs a `SELECT DISTINCT` on the remote database and answers
Expand Down
1 change: 1 addition & 0 deletions packages/drivers/driver-turso/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
"devDependencies": {
"@types/node": "^26.6.3",
"better-sqlite3": "^13.0.3",
"tsx": "^4.23.12",
"typescript": "^6.0.3",
"vitest": "^4.1.11"
},
Expand Down
12 changes: 11 additions & 1 deletion packages/drivers/driver-turso/src/remote-transport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1595,11 +1595,21 @@ export class RemoteTransport {
return rows[0] ?? null;
}

/**
* `insertOnlyColumns` — columns written on the INSERT leg and left alone on
* the MERGE leg, so a value the caller (or the driver, on its behalf) put on
* the row for a NEW row never overwrites the one an existing row already
* holds. The transport is told WHICH columns, not WHY: the driver decides
* from the schema it holds and this class does not (the same division
* `setFilterColumnSql` and `setTenantFieldResolver` draw). A column named
* here that is also a merge key is simply a merge key.
*/
async upsert(
object: string,
data: Record<string, unknown>,
conflictKeys?: string[],
table: string = object,
insertOnlyColumns: readonly string[] = [],
): Promise<Record<string, unknown>> {
await this.ensureConnected();

Expand All @@ -1618,7 +1628,7 @@ export class RemoteTransport {
const mergeKeys = conflictKeys && conflictKeys.length > 0 ? conflictKeys : ['id'];

// Build ON CONFLICT ... DO UPDATE SET
const updateCols = columns.filter((c) => !mergeKeys.includes(c));
const updateCols = columns.filter((c) => !mergeKeys.includes(c) && !insertOnlyColumns.includes(c));
const updateClauses = updateCols.map((col) => `"${col}" = excluded."${col}"`).join(', ');

let sql = `INSERT INTO ${this.tableSql(table)} (${columns.map((c) => `"${c}"`).join(', ')}) VALUES (${placeholders})`;
Expand Down
Loading
Loading