Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: update sync plugin to work with the latest Electric sync server #531

Merged
merged 3 commits into from
Feb 12, 2025
Merged

fix: update sync plugin to work with the latest Electric sync server #531

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f5801fb
update sync plugin
samwillis Feb 11, 2025
16a35f2
docs stylecheck
Feb 11, 2025
129ff7d
Fit types in tests
samwillis Feb 12, 2025
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/gentle-ways-knock.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@electric-sql/pglite-sync': patch
---

Update the sync plugin to work with the latest Electric sync server
3 changes: 2 additions & 1 deletion docs/docs/api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,9 +362,10 @@ await pg.describeQuery('SELECT * FROM test WHERE name = $1', ['test'])
```

### clone

`.clone(): Promise<PGlite>`

Clones the current instance. This is useful when a series of operations, like unit or integration test, need to be run on the same database without having to recreate the database each time, or for each test.
Clones the current instance. This is useful when a series of operations, like unit or integration test, need to be run on the same database without having to recreate the database each time, or for each test.

## Properties

Expand Down
2 changes: 1 addition & 1 deletion packages/pglite-sync/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@
"dist"
],
"dependencies": {
"@electric-sql/client": "~0.9.0"
"@electric-sql/client": "1.0.0-beta.3"
},
"devDependencies": {
"@electric-sql/pglite": "workspace:*",
Expand Down
23 changes: 18 additions & 5 deletions packages/pglite-sync/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import type { Offset, ShapeStreamOptions } from '@electric-sql/client'
import type { Offset, Row, ShapeStreamOptions } from '@electric-sql/client'
import {
ChangeMessage,
ShapeStream,
Expand All @@ -12,6 +12,10 @@ import type {
Transaction,
} from '@electric-sql/pglite'

interface LegacyChangeMessage<T extends Row<unknown>> extends ChangeMessage<T> {
offset?: Offset
}

export type MapColumnsMap = Record<string, string>
export type MapColumnsFn = (message: ChangeMessage<any>) => Record<string, any>
export type MapColumns = MapColumnsMap | MapColumnsFn
Expand Down Expand Up @@ -154,7 +158,7 @@ async function createPlugin(
// _very_ large shapes - either we should commit batches to
// a temporary table and copy over the transactional result
// or use a separate connection to hold a long transaction
let messageAggregator: ChangeMessage<any>[] = []
let messageAggregator: LegacyChangeMessage<any>[] = []
let truncateNeeded = false
// let lastLSN: string | null = null // Removed until Electric has stabilised on LSN metadata
let lastCommitAt: number = 0
Expand Down Expand Up @@ -248,8 +252,9 @@ async function createPlugin(
metadataSchema,
shapeKey: options.shapeKey,
shapeId: shapeHandle,
lastOffset:
messageAggregator[messageAggregator.length - 1].offset,
lastOffset: getMessageOffset(
messageAggregator[messageAggregator.length - 1],
),
})
}
})
Expand Down Expand Up @@ -359,7 +364,7 @@ async function createPlugin(
return stream.isUpToDate
},
get shapeId() {
return stream.shapeHandle
return stream.shapeHandle!
},
stream,
subscribe: (cb: () => void, error: (err: Error) => void) => {
Expand Down Expand Up @@ -667,3 +672,11 @@ function subscriptionMetadataTableName(metadatSchema: string) {
}

const subscriptionTableName = `shape_subscriptions_metadata`

function getMessageOffset(message: LegacyChangeMessage<any>): Offset {
if (message.offset) {
return message.offset
} else {
return `${message.headers.lsn}_${message.headers.op_position}` as Offset
}
}
25 changes: 0 additions & 25 deletions packages/pglite-sync/test/sync.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,6 @@ describe('pglite-sync', () => {
// insert
await feedMessage({
headers: { operation: 'insert' },
offset: '-1',
key: 'id1',
value: {
id: 1,
Expand All @@ -83,7 +82,6 @@ describe('pglite-sync', () => {
// update
await feedMessage({
headers: { operation: 'update' },
offset: '-1',
key: 'id1',
value: {
id: 1,
Expand All @@ -102,7 +100,6 @@ describe('pglite-sync', () => {
// delete
await feedMessage({
headers: { operation: 'delete' },
offset: '-1',
key: 'id1',
value: {
id: 1,
Expand Down Expand Up @@ -315,7 +312,6 @@ describe('pglite-sync', () => {
await feedMessages([
{
headers: { operation: 'insert' },
offset: `1_${numInserts}`,
key: `id${numInserts}`,
value: {
id: numInserts,
Expand All @@ -326,7 +322,6 @@ describe('pglite-sync', () => {
{ headers: { control: 'must-refetch' } },
{
headers: { operation: 'insert' },
offset: `2_1`,
key: `id21`,
value: {
id: 21,
Expand Down Expand Up @@ -466,7 +461,6 @@ describe('pglite-sync', () => {
// insert
await feedMessage({
headers: { operation: 'insert' },
offset: '-1',
key: 'id1',
value: {
id: 1,
Expand All @@ -485,7 +479,6 @@ describe('pglite-sync', () => {
// update with no columns to update
await feedMessage({
headers: { operation: 'update' },
offset: '-1',
key: 'id1',
value: {
id: 1,
Expand Down Expand Up @@ -555,7 +548,6 @@ describe('pglite-sync', () => {

await feedMessage({
headers: { operation: 'insert' },
offset: '-1',
key: 'id1',
value: {
id: 'id1',
Expand Down Expand Up @@ -619,7 +611,6 @@ describe('pglite-sync', () => {
),
{
headers: { operation: 'update' as const },
offset: `1_${numInserts}`,
key: `id0`,
value: {
id: 0,
Expand Down Expand Up @@ -683,7 +674,6 @@ describe('pglite-sync', () => {
const specialCharMessages: Message[] = [
{
headers: { operation: 'insert' },
offset: '1_0',
key: 'id1',
value: {
id: 1,
Expand All @@ -693,7 +683,6 @@ describe('pglite-sync', () => {
},
{
headers: { operation: 'insert' },
offset: '2_0',
key: 'id2',
value: {
id: 2,
Expand All @@ -703,7 +692,6 @@ describe('pglite-sync', () => {
},
{
headers: { operation: 'insert' },
offset: '3_0',
key: 'id3',
value: {
id: 3,
Expand Down Expand Up @@ -784,7 +772,6 @@ describe('pglite-sync', () => {
(_, idx) =>
({
headers: { operation: 'insert' },
offset: `1_${idx}`,
key: `id${idx}`,
value: {
id: idx,
Expand Down Expand Up @@ -872,7 +859,6 @@ describe('pglite-sync', () => {
// await feedMessages([
// {
// headers: { operation: 'insert' },
// offset: '1_1', // Transaction 1
// key: 'id1',
// value: {
// id: 1,
Expand All @@ -882,7 +868,6 @@ describe('pglite-sync', () => {
// },
// {
// headers: { operation: 'insert' },
// offset: '1_2', // Same transaction
// key: 'id2',
// value: {
// id: 2,
Expand All @@ -892,7 +877,6 @@ describe('pglite-sync', () => {
// },
// {
// headers: { operation: 'insert' },
// offset: '2_1', // New transaction
// key: 'id3',
// value: {
// id: 3,
Expand Down Expand Up @@ -974,19 +958,16 @@ describe('pglite-sync', () => {
await feedMessages([
{
headers: { operation: 'insert' },
offset: '1_1',
key: 'id1',
value: { id: 1, task: 'task1', done: false },
},
{
headers: { operation: 'insert' },
offset: '2_1',
key: 'id2',
value: { id: 2, task: 'task2', done: false },
},
{
headers: { operation: 'insert' },
offset: '3_1',
key: 'id3',
value: { id: 3, task: 'task3', done: false },
},
Expand Down Expand Up @@ -1054,19 +1035,16 @@ describe('pglite-sync', () => {
await feedMessages([
{
headers: { operation: 'insert' },
offset: '1_1',
key: 'id1',
value: { id: 1, task: 'task1', done: false },
},
{
headers: { operation: 'insert' },
offset: '1_2',
key: 'id2',
value: { id: 2, task: 'task2', done: false },
},
{
headers: { operation: 'insert' },
offset: '1_3',
key: 'id3',
value: { id: 3, task: 'task3', done: false },
},
Expand Down Expand Up @@ -1230,7 +1208,6 @@ describe('pglite-sync', () => {
await feedMessages([
{
headers: { operation: 'insert' },
offset: '1_1',
key: 'id1',
value: {
id: 1,
Expand All @@ -1240,7 +1217,6 @@ describe('pglite-sync', () => {
},
{
headers: { operation: 'insert' },
offset: '1_2',
key: 'id2',
value: {
id: 2,
Expand All @@ -1257,7 +1233,6 @@ describe('pglite-sync', () => {
await feedMessages([
{
headers: { operation: 'insert' },
offset: '1_3',
key: 'id3',
value: {
id: 3,
Expand Down
11 changes: 6 additions & 5 deletions pnpm-lock.yaml
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.