live queries #1408
live queries #1408
Conversation
packages/client/src/index.ts
Outdated
|
|
||
| sse.onmessage = (event) => { | ||
| // @ts-ignore | ||
| callback(prepared.mapResult(JSON.parse(event.data), true).rows); |
There was a problem hiding this comment.
Any error handling to consider here as well?
| } else { | ||
| return client.live( | ||
| (db) => db.select().from(status), | ||
| () => { | ||
| _query(client.db).then(onData).catch(onError); | ||
| }, | ||
| onError, | ||
| ); |
There was a problem hiding this comment.
Slightly strange block lol but I follow
|
|
||
| build.initNamespace({ isSchemaRequired: false }); |
| if (listenConnection) { | ||
| if (listenConnection.dialect === "postgres") { | ||
| listenConnection.connection.release(); | ||
| } | ||
| } |
There was a problem hiding this comment.
Why is listenConnection not a property of database ? They seem to be passed around together.
| } catch (error) { | ||
| return c.text((error as Error).message, 500); | ||
| } |
There was a problem hiding this comment.
May want to strip some of the details from this error object before returning, but can add this after prerelease.
| }), | ||
| }); | ||
| } | ||
| statusResult = await statusResolver.promise; |
There was a problem hiding this comment.
This waits for the next status update?
| export const client = ({ db }: { db: ReadonlyDrizzle<Schema> }) => { | ||
| // @ts-ignore | ||
| const session: PgSession = db._.session; |
There was a problem hiding this comment.
Does this mean all queries will use the same connection? Seems important to use a pool here.
| await sql | ||
| .raw(` | ||
| CREATE OR REPLACE FUNCTION ${notification} | ||
| RETURNS trigger | ||
| LANGUAGE plpgsql | ||
| AS $$ | ||
| BEGIN | ||
| NOTIFY ${channel}; | ||
| RETURN NULL; | ||
| END; | ||
| $$;`) | ||
| .execute(qb.internal); | ||
|
|
||
| await sql | ||
| .raw(` | ||
| CREATE OR REPLACE TRIGGER ${trigger} | ||
| AFTER INSERT OR UPDATE OR DELETE | ||
| ON "${preBuild.namespace}"._ponder_status | ||
| FOR EACH STATEMENT | ||
| EXECUTE PROCEDURE ${notification};`) | ||
| .execute(qb.internal); | ||
| }); |
There was a problem hiding this comment.
My instinct is that these should happen in the same spot that we create the _ponder_status table (if not exists).
| if (dialect === "postgres") { | ||
| return { | ||
| dialect: "postgres", | ||
| connection: await (driver as PostgresDriver).internal.connect(), |
There was a problem hiding this comment.
What happens if this connection drops?
No description provided.