Skip to content

Commit

Permalink
WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
@samwillis
samwillis committed Jul 29, 2024
1 parent 2f0af8f commit 912a1c2
Show file tree
Hide file tree
Showing 3 changed files with 116 additions and 43 deletions.
2 changes: 1 addition & 1 deletion packages/pglite/src/extensionUtils.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ export async function loadExtensionBundle(
const response = await fetch(bundlePath.toString());
if (!response.ok || !response.body) {
return null;
} else if (response.headers.get('Content-Encoding') === 'gzip') {
} else if (response.headers.get("Content-Encoding") === "gzip") {
// Although the bundle is manually compressed, some servers will recognize
// that and add a content-encoding header. Fetch will then automatically
// decompress the response.
Expand Down
40 changes: 20 additions & 20 deletions packages/pglite/src/live/index.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
async query<T>(
query: string,
params: any[] | undefined | null,
callback: (results: Results<T>) => void
callback: (results: Results<T>) => void,
) {
const id = liveQueryCounter++;

Expand All @@ -35,7 +35,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
// Create a temporary view with the query
await tx.query(
`CREATE OR REPLACE TEMP VIEW live_query_${id}_view AS ${query}`,
params ?? []
params ?? [],
);

// Get the tables used in the view and add triggers to notify when they change
Expand Down Expand Up @@ -65,7 +65,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
`table_change__${table.schema_name}__${table.table_name}`,
async () => {
refresh();
}
},
);
unsubList.push(unsub);
}
Expand Down Expand Up @@ -96,7 +96,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
query: string,
params: any[] | undefined | null,
key: string,
callback: (changes: Array<Change<T>>) => void
callback: (changes: Array<Change<T>>) => void,
) {
const id = liveQueryCounter++;
let tables: { table_name: string; schema_name: string }[];
Expand All @@ -107,7 +107,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
// Create a temporary view with the query
await tx.query(
`CREATE OR REPLACE TEMP VIEW live_query_${id}_view AS ${query}`,
params ?? []
params ?? [],
);

// Get the tables used in the view and add triggers to notify when they change
Expand Down Expand Up @@ -147,7 +147,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
${columns
.map(
({ column_name }) =>
`curr."${column_name}" AS "${column_name}"`
`curr."${column_name}" AS "${column_name}"`,
)
.join(",\n")},
ARRAY[]::text[] AS __changed_columns__
Expand Down Expand Up @@ -183,7 +183,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
WHEN curr."${column_name}" IS DISTINCT FROM prev."${column_name}"
THEN curr."${column_name}"
ELSE NULL::${data_type}
END AS "${column_name}"`
END AS "${column_name}"`,
)
.join(",\n")},
ARRAY(SELECT unnest FROM unnest(ARRAY[${columns
Expand All @@ -194,10 +194,10 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
WHEN curr."${column_name}" IS DISTINCT FROM prev."${column_name}"
THEN '${column_name}'
ELSE NULL
END`
END`,
)
.join(
", "
", ",
)}]) WHERE unnest IS NOT NULL) AS __changed_columns__
FROM curr
INNER JOIN prev ON curr.${key} = prev.${key}
Expand All @@ -219,7 +219,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {

// Get the changes
changes = await tx.query<any>(
`EXECUTE live_query_${id}_diff${stateSwitch};`
`EXECUTE live_query_${id}_diff${stateSwitch};`,
);
});

Expand All @@ -236,7 +236,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
`table_change__${table.schema_name}__${table.table_name}`,
async () => {
refresh();
}
},
);
unsubList.push(unsub);
}
Expand All @@ -261,7 +261,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
// Fields
const fields = changes!.fields.filter(
(field) =>
!["__after__", "__op__", "__changed_columns__"].includes(field.name)
!["__after__", "__op__", "__changed_columns__"].includes(field.name),
);

// Return the initial results
Expand All @@ -277,7 +277,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
query: string,
params: any[] | undefined | null,
key: string,
callback: (results: Results<Change<T>>) => void
callback: (results: Results<Change<T>>) => void,
) {
const rowsMap: Map<any, any> = new Map();
const afterMap: Map<any, any> = new Map();
Expand Down Expand Up @@ -340,7 +340,7 @@ const setup = async (pg: PGliteInterface, emscriptenOpts: any) => {
fields,
});
}
}
},
);

firstRun = false;
Expand Down Expand Up @@ -378,7 +378,7 @@ export const live = {
*/
async function getTablesForView(
tx: Transaction | PGliteInterface,
viewName: string
viewName: string,
): Promise<{ table_name: string; schema_name: string }[]> {
return (
await tx.query<{
Expand All @@ -399,7 +399,7 @@ async function getTablesForView(
)
AND d.deptype = 'n';
`,
[viewName]
[viewName],
)
).rows.filter((row) => row.table_name !== viewName);
}
Expand All @@ -412,14 +412,14 @@ async function getTablesForView(
async function addNotifyTriggersToTables(
tx: Transaction | PGliteInterface,
tables: { table_name: string; schema_name: string }[],
tableNotifyTriggersAdded: Set<string>
tableNotifyTriggersAdded: Set<string>,
) {
const triggers = tables
.filter(
(table) =>
!tableNotifyTriggersAdded.has(
`${table.schema_name}_${table.table_name}`
)
`${table.schema_name}_${table.table_name}`,
),
)
.map((table) => {
return `
Expand All @@ -439,6 +439,6 @@ async function addNotifyTriggersToTables(
await tx.exec(triggers);
}
tables.map((table) =>
tableNotifyTriggersAdded.add(`${table.schema_name}_${table.table_name}`)
tableNotifyTriggersAdded.add(`${table.schema_name}_${table.table_name}`),
);
}
Loading

0 comments on commit 912a1c2

Please sign in to comment.