Syncing from a database.

Some data never leaves the system that owns it as a file or an API response. An ERP or a legacy application keeps it in its own database, and your application can connect to that database but should not write to it.

A DatabaseSyncDefinition reads a query on that connection on a schedule. Each fetch reads the query in pages ordered by a key and writes every row to one NDJSON file, called a snapshot, in the staging area. The snapshot then goes through the same pipeline as a file feed: the checksum ledger, chunked writes, row failures, archiving, sync:status, and sync:retry.

namespace App\Syncs;

use App\Models\Order;
use ByRcsc\LaravelDataSync\Definitions\DatabaseSyncDefinition;
use ByRcsc\LaravelDataSync\Definitions\Field;
use ByRcsc\LaravelDataSync\Definitions\Watermark;
use Illuminate\Console\Scheduling\Event;
use Illuminate\Database\Query\Builder;

final class LegacyOrdersSync extends DatabaseSyncDefinition
{
    public function connection(): string
    {
        return 'erp';
    }

    public function query(Builder $query): Builder
    {
        return $query->from('Orders')
            ->select(['OrderNo', 'Customer', 'Total', 'UpdatedAt']);
    }

    public function keyColumns(): array
    {
        return ['OrderNo'];
    }

    public function watermark(): ?Watermark
    {
        return Watermark::timestamp('UpdatedAt', overlapSeconds: 300);
    }

    public function model(): string
    {
        return Order::class;
    }

    public function fields(): array
    {
        return [
            Field::make('order_no', from: 'OrderNo'),
            Field::make('customer', from: 'Customer'),
            Field::make('total', from: 'Total'),
        ];
    }

    public function matchOn(): array
    {
        return ['order_no'];
    }

    public function schedule(Event $event): void
    {
        $event->everyFifteenMinutes();
    }
}

Scaffold one with:

php artisan make:sync LegacyOrdersSync --database

Add the erp connection to config/database.php, register the class in data-sync.syncs, then run it with php artisan sync:run legacy-orders.

What changes from a file sync

DatabaseSyncDefinition extends SyncDefinition. Field mapping, matchOn(), rules(), the write hooks, chunking, atomic(), maxErrors(), payload capture, queue routing, and scheduling work the same way as for a file feed.

Two methods are fixed. format() always returns Format::ndjson(), because the package writes the snapshot itself. source() always points at the definition's directory in the staging area. Both methods are final.

The definition gains these methods:

The package validates these methods when it resolves the definition, without opening a connection, so your application still boots while the source database is down:

• connection() names a connection that exists in database.connections.
• keyColumns() returns at least one column, and no empty names.
• query() reads from a table, view, or subquery, and sets no limit or offset.
• A Watermark::rowVersion() watermark is on a sqlsrv connection.
• pageSize() and lockTimeout() return 1 or more, and overlapSeconds is not negative.

A failure throws an InvalidDefinitionException that names the method to fix.

Write the query

query() receives an empty query builder on the definition's connection. Choose the table and the columns, and add any filter:

public function query(Builder $query): Builder
{
    return $query->from('Orders')
        ->select(['OrderNo', 'Customer', 'Total', 'UpdatedAt'])
        ->where('Status', '!=', 'draft');
}

Return the base query only. The package adds the watermark window, replaces any ordering with keyColumns(), and pages the result. A query that sets its own limit() or offset() fails validation.

Select every column that a field maps and every key column. The watermark column can stay out of the select, because the package filters on it in the query and never needs its value in a row. Each column's select alias becomes a key in the snapshot, so map fields by alias, for example Field::make('order_no', from: 'OrderNo'). The NDJSON reader turns every value into a string or null. Map a JSON column to an attribute with a JSON cast to store it as structured JSON. See JSON attributes.

A query can join tables or union several queries. It runs on one connection, so it cannot join a table on another connection.

Choose the key columns

keyColumns() lists the source columns that put rows in one fixed order. The fetch reads the first pageSize() rows in that order, then asks for the rows after the last key it read, and so on. Each page is one bounded query, so a large table never needs a large OFFSET.

List every column of a composite key:

public function keyColumns(): array
{
    return ['DocEntry', 'LineNum'];
}

The key must be unique and never null. A row with a null key column fails the fetch. A qualified name such as orders.id is allowed in a joined query, and the row must return the column as id.

Pick a watermark

A watermark tells the fetch which rows changed since the last completed run, so a run reads those rows instead of the whole table. Pick it by what the source table has.

No watermark

Every run reads the whole query. An unchanged table produces the same snapshot, and the ledger skips it, so the run writes nothing. A table that changed produces a new snapshot, and the upsert on matchOn() updates the rows you already have.

A timestamp column

public function watermark(): ?Watermark
{
    return Watermark::timestamp('UpdatedAt', overlapSeconds: 300);
}

Before it reads, the fetch takes MAX(UpdatedAt) as the upper bound. It then reads the rows above the stored watermark and up to that bound. A row that changes during the read gets a later time, so the next run reads it.

A transaction that started before the last fetch can commit an older UpdatedAt after the fetch read past it. overlapSeconds re-reads that many seconds below the stored watermark to catch those rows. The upsert on matchOn() makes the re-read rows harmless. Set the overlap above your longest write transaction on the source.

The first run reads every row, including rows with no value in the column yet. The package stores the bound exactly as the driver returned it, so a SQL Server datetime2 keeps all seven fractional digits.

A SQL Server rowversion column

public function watermark(): ?Watermark
{
    return Watermark::rowVersion('RowVer');
}

SQL Server changes a rowversion column on every write to the row. The fetch takes MIN_ACTIVE_ROWVERSION() as the upper bound, so a run reads committed rows only. A row that commits after the read lands in the next run, with no overlap needed.

The bound moves with every write anywhere in the database. A run over a table that did not change is therefore a completed zero-row run, not a skipped one.

Watermark::rowVersion() works only on a sqlsrv connection. Leave the rowversion column out of the select in query(). Its value is binary, and a snapshot cannot store it.

Change the watermark

The package stores the mode and the column with the watermark. If you change either one, the next fetch fails with an error that names the sync_watermarks row to delete. Delete it, and the next run reads every row again.

When the watermark moves

The package moves the watermark only when a run finishes as completed. The rules are the same as for an API sync:

When a run does not write every row, the next fetch reads the same window again, plus any rows that changed since. If the same row fails on every run, the watermark never moves and the window keeps growing. When a definition's last 5 runs all finish without completed, sync:status and sync:doctor warn about it. data-sync.stalled_watermark_runs sets the count. List the failed rows with sync:status <name> --failures, fix the cause, and replay them with sync:retry.

The source must match the definition

Before it reads any rows, the fetch checks the source table's columns. Every mapped column, every key column, and the watermark column must exist, and the watermark column must have a type that its mode can compare. A mismatch fails the fetch with an error that names each column to fix. A query with joins or unions is checked against its first page of rows instead.

A binary column in the select, such as varbinary, fails the run with the column's name, because a snapshot cannot store raw bytes. Drop the column, or on SQL Server select it as text with CONVERT(varchar(max), column, 1).

sync:doctor runs the same table check with one schema read per definition. It also reports a connection it cannot reach and a missing PDO extension, with the command that installs it.

Read without writing

The package only reads from the source connection. It records what it has synced in your application's database: the watermark, the upsert on matchOn(), and the checksum ledger. It never marks rows in the source.

Give the connection a read-only login. On SQL Server, turn on READ_COMMITTED_SNAPSHOT for the database, or read from a readable replica, so the fetch neither blocks writers nor waits for them.

Do not add WITH (NOLOCK) to the query. A NOLOCK read can return a row twice or skip it while SQL Server splits a page.

To mark rows on the source side after a sync, write to the source from a SyncCompleted listener in your application. See send acknowledgements back for the pattern.

Run modes

sync:run legacy-orders fetches in the command's process, then dispatches the snapshot to the queue in the same way as a discovered file. The scheduler does the same. With --now, the command also processes the snapshot in its own process.

--dry-run reads the source and writes a snapshot, then counts rows without writing them. It does not move the watermark.

A fetch takes the same per-definition cache lock as file discovery. A scheduled fetch that starts while another fetch of the same definition is running reports Skipped [legacy-orders] because discovery is already running. and exits zero. The lock expires after lockTimeout() seconds, 300 by default. A first read of a large table can take longer, so raise lockTimeout() above the longest fetch you expect. Otherwise the scheduler can start a second fetch of the same rows.

The scheduler runs due commands one after another, and a fetch reads its rows inside the scheduled command. With several slow sources, call $event->runInBackground() in schedule().

When a fetch fails

When the query fails, for example because the source is down:

• The package records a failed run and dispatches SyncFailed. The run's error_summary names the connection and the driver's error.
• The rows read before the failure become the snapshot, and the package copies it to the failed area.
• The watermark does not move.
• sync:run prints the error and exits non-zero.

The next scheduled fetch reads from the same watermark, and that is usually the only recovery a failed fetch needs.

Limits

• Deleted rows stay. A watermark only sees rows that still exist, so the package never reads a row that the source deleted, and never deletes the model it wrote for that row.
• One connection per query. A query cannot join tables on two connections.
• Package tables on SQL Server are untested. A definition can read from SQL Server. The package tests its own ledger and history tables on SQLite, MySQL, and PostgreSQL only.
• The fetch finishes before processing starts. The package spills the snapshot to a temporary file as it grows, and processing begins after the last page.
• No writes to the source. See read without writing.

What to read next

• Definition providers to register one database definition for each source an account connects.
• Field mapping for mapping source columns to model attributes.
• Testing for running a database sync against SQLite in a test.