debug6

.github/workflows/deploy.yml

@@ -2,7 +2,7 @@ name: Deploy
 on:
   push:
     branches:
-      - main
+      - sven/debug6
   repository_dispatch:
 jobs:
   deploy:
@@ -257,7 +257,8 @@ jobs:
             echo -e "name = \"wildebeest-consumer-${{ env.NAME_SUFFIX }}\"\n" >> consumer/wrangler.toml
 
             echo -e "[[queues.consumers]]\n" >> consumer/wrangler.toml
-            echo -e "max_batch_size = 10\n" >> consumer/wrangler.toml
+            echo -e "max_batch_size = 1\n" >> consumer/wrangler.toml
+            echo -e "max_concurrency = 1\n" >> consumer/wrangler.toml
             echo -e "max_batch_timeout = 30\n" >> consumer/wrangler.toml
             echo -e "max_retries = 10\n" >> consumer/wrangler.toml
             echo -e "queue = \"wildebeest-${{ env.NAME_SUFFIX }}\"\n" >> consumer/wrangler.toml

backend/src/database/d1.ts

@@ -35,6 +35,7 @@ export default function make({ DATABASE }: Pick<Env, 'DATABASE'>): Database {
 	const db = DATABASE as any
 	db.qb = qb
 	db.client = 'd1'
+	db.close = async () => {}
 
 	return db as Database
 }

backend/src/database/index.ts

@@ -12,6 +12,7 @@ export interface Result<T = unknown> {
 export interface Database {
 	prepare(query: string): PreparedStatement
 	dump(): Promise<ArrayBuffer>
+	close(): Promise<void>
 	batch<T = unknown>(statements: PreparedStatement[]): Promise<Result<T>[]>
 	exec<T = unknown>(query: string): Promise<Result<T>>
 	qb: QueryBuilder
@@ -37,9 +38,9 @@ export interface QueryBuilder {
 }
 
 export async function getDatabase(env: Pick<Env, 'DATABASE' | 'NEON_DATABASE_URL'>): Promise<Database> {
-	if (env.NEON_DATABASE_URL !== undefined) {
-		return neon(env)
-	}
+	// if (env.NEON_DATABASE_URL !== undefined) {
+	// 	return neon(env)
+	// }
 
 	return d1(env)
 }

backend/src/database/neon.ts

@@ -41,13 +41,17 @@ const qb: QueryBuilder = {
 }
 
 export default async function make(env: Pick<Env, 'NEON_DATABASE_URL'>): Promise<Database> {
-	const client = new neon.Client(env.NEON_DATABASE_URL)
+	const client = new neon.Pool({ connectionString: env.NEON_DATABASE_URL })
 	await client.connect()
 
 	return {
 		client: 'neon',
 		qb,
 
+		async close() {
+			await client.end()
+		},
+
 		prepare(query: string) {
 			return new PreparedStatement(env, query, [], client)
 		},
@@ -82,11 +86,11 @@ export default async function make(env: Pick<Env, 'NEON_DATABASE_URL'>): Promise
 
 export class PreparedStatement {
 	private env: Pick<Env, 'NEON_DATABASE_URL'>
-	private client: neon.Client
+	private client: neon.Pool
 	public query: string
 	public values: any[]
 
-	constructor(env: Pick<Env, 'NEON_DATABASE_URL'>, query: string, values: any[], client: neon.Client) {
+	constructor(env: Pick<Env, 'NEON_DATABASE_URL'>, query: string, values: any[], client: neon.Pool) {
 		this.env = env
 		this.query = query
 		this.values = values
@@ -103,7 +107,9 @@ export class PreparedStatement {
 		}
 		const query = sqliteToPsql(this.query)
 
+		console.log({ query })
 		const results = await this.client.query(query, this.values)
+		console.log('res')
 		if (results.rows.length !== 1) {
 			throw new Error(`expected a single row, returned ${results.rows.length} row(s)`)
 		}
@@ -117,7 +123,9 @@ export class PreparedStatement {
 
 	async all<T = unknown>(): Promise<Result<T>> {
 		const query = sqliteToPsql(this.query)
+		console.log({ query })
 		const results = await this.client.query(query, this.values)
+		console.log('res')
 
 		return {
 			results: results.rows as T[],

consumer/src/inbox.ts

@@ -17,10 +17,12 @@ export async function handleInboxMessage(env: Env, actor: Actor, message: InboxM
 
 	await activityHandler.handle(domain, activity, db, message.userKEK, adminEmail, message.vapidKeys)
 
+	console.log('generate timeline')
 	// Assuming we received new posts or a like, pregenerate the user's timelines
 	// and notifications.
 	await Promise.all([
 		timeline.pregenerateTimelines(domain, db, cache, actor),
 		notification.pregenerateNotifications(db, cache, actor, domain),
 	])
+	console.log('done')
 }

consumer/src/index.ts

@@ -23,6 +23,7 @@ export default {
 	async queue(batch: MessageBatch<MessageBody>, env: Env, ctx: ExecutionContext) {
 		const sentry = initSentryQueue(env, ctx)
 		const db = await getDatabase(env)
+		console.log('batch of ' + batch.messages.length + ' messages')
 
 		try {
 			for (const message of batch.messages) {
@@ -51,5 +52,8 @@ export default {
 			}
 			console.error(err.stack, err.cause)
 		}
+
+		await db.close()
+		// FIXME: runtime fails after here to close the WS. maybe it's not closed correclty?
 	},
 }

frontend/src/routes/(admin)/settings/(admin)/server-settings/layout.tsx

@@ -11,7 +11,6 @@ export type ServerSettingsData = ServerBrandingData & ServerAboutData
 
 export const serverSettingsLoader = loader$<Promise<Partial<ServerSettingsData>>>(async ({ platform }) => {
 	const database = await getDatabase(platform)
-
 	const settings = await getSettings(database)
 
 	return JSON.parse(JSON.stringify(settings))

tf/main.tf

@@ -101,7 +101,7 @@ resource "random_password" "user_key" {
 resource "cloudflare_pages_project" "wildebeest_pages_project" {
   account_id = var.cloudflare_account_id
   name              = "wildebeest-${lower(var.name_suffix)}"
-  production_branch = "main"
+  production_branch = "sven/debug6"
 
   deployment_configs {
     production {