Forem: Keith Mifsud

Nuxt & Cloudflare Queues: Building a Data Sync Pipeline using Vectorize

Keith Mifsud — Tue, 30 Dec 2025 06:52:51 +0000

Continuing from the last article, we’ve already deployed our AI-powered real estate listing demo Nuxt application to Cloudflare, including the D1 database and AI Workers binding. In this article, we will set up Cloudflare Vectorize and implement a Cloudflare Queue to populate the Vector store with embeddings for our locations.

This article is the second part of the Nuxt & Cloudflare AI Vector Pipeline Series, a three-part series.

Part one - Nuxt & Cloudflare Vectorize: Setting up D1, Drizzle, and Workers AI
Part two - Nuxt & Cloudflare Queues: Building a Data Sync Pipeline using Vectorize
Part three - Implementing semantic matching in Nuxt with Cloudflare Vectorize

The code for the entire demo application is publicly available on GitHub: Nuxt & Cloudflare AI Vector Pipeline Series - give it a ⭐.

Create a vector index on Cloudflare Vectorize and bind it to our Nuxt application

As briefly explained in the previous article, Vectorize is Cloudflare’s native vector store. A vector store is essentially a database designed to hold text embeddings and index them for fast querying. Text embeddings are numeric representations (lists of numbers) that AI models use to read and understand data. Feel free to ask in comments below if my subtle explanation of Vectorize and vector stores leads to more questions - I know it's too broad 😆.

Cloudflare’s Workers AI supports several embedding models; we will use @cf/baai/bge-m3 because of my experience with it on similar projects and its multilingual support, which is ideal for semantic matching use cases.

We will use Wrangler CLI to create the vector index on Cloudflare and bind it to the project through Wrangler’s config so we can build the index with the locations’ text embeddings later.

To use the bge-m3 embedding model, we need to specify a dimension of 1024. Run:

npx wrangler vectorize create locations-index --dimensions=1024 --metric=cosine

and add the bindings in /wrangler.toml:

[[vectorize]]
binding = "VECTORIZE"
index_name = "locations-index"

Now that we've created the location vector index, we need to enable filtering on our metadata. While upserts work automatically based on ID, we must explicitly index locationId so we can filter search results or delete obsolete records by their location ID in the future.

npx wrangler vectorize create-metadata-index locations-index --property-name=locationId --type=string

We can now build our code and deploy it to make sure Vectorize is being used, and the index is created:

pnpm run build
npx wrangler deploy

The CLI output should show that our worker has access to Vectorize and AI (which we previously added):

Your Worker has access to the following bindings:
Binding                               Resource             
env.DB (property-sync-db)             D1 Database          
env.VECTORIZE (locations-index)       Vectorize Index      
env.ASSETS                            Assets
env.AI                                AI

Upserting vectors using Cloudflare Workers AI and Vectorize

We will use a simple repository pattern for Cloudflare Vectorize, allowing us to easily swap implementations if needed. This sounds like overkill for such a small demo application, right? I recently avoided a substantial amount of time wasted when NuxtHub announced they’re deprecating their AI features. I was building a similar feature using hubAi() and hubVectorize() from NuxtHub’s core module, but I had to refactor it before I finished. Thanks to my use of the repository pattern, I only had to swap the repository implementation from NuxthubVectorAIRepository to CloudflareVectorAIRepository as, all other code was bound to the VectorAIRepository interface.

💡 I don’t see anything wrong with enforcing a Repository Pattern throughout the application. I also don’t see anything wrong with using the pattern for only some specific dependencies. In this case. I know that there’s a chance we will need to change the Vector and AI implementations, but changing the ORM (Drizzle) implementation is less likely. Let me know in the comments below your thoughts on the Repository Pattern in Nuxt.

I will go into detail about Queues and Handlers in the next section of this article. For now, we just need to know that the vectorisation of the location data will be queued, and that a queue handler will depend on a Location Vector Repository. Let’s add this dependency.

Create /server/utils/queueHandlers/repositories/LocationVectorRepository.ts with:

import { Location } from '~~/server/database/types'

export interface LocationVectorRepository {

  /**
   * Upserts locations into the Vector Store.
   *
   * @param locations - An array of Location objects to be upserted.
   * @returns A promise that resolves to the number of locations upserted.
   */
  upsertLocations (
    locations: Location[],
  ): Promise<number>;
}

And install the following types from Cloudflare Workers if you haven’t already: pnpm install @cloudflare/workers-types.

Database types in /server/database/types.ts:

import * as schema from './schema';

export type Location = typeof schema.locations.$inferSelect;

export type Agent = typeof schema.agents.$inferSelect;

export type Property = typeof schema.properties.$inferSelect;

AI types in /server/types/ai-models.ts:

export interface BGEEmbeddingResponse {
  data?: number[][];
  response?: number[][];
  shape?: number[];
}

We can now finally implement the upsertLocations() method in our Cloudflare implementation of the LocationVectorRepository interface.

Let’s start by creating the repository class in /server/repositories/cloudflare/CloudflareVectorAIRepository.ts and add the following contents:

import {
  LocationVectorRepository,
} from '~~/server/utils/queueHandlers/repositories/LocationVectorRepository'
import { Location } from '~~/server/database/types'
import type { Ai, VectorizeIndex } from '@cloudflare/workers-types'
import { BGEEmbeddingResponse } from '~~/server/types/ai-models'

export class CloudflareVectorAIRepository implements LocationVectorRepository {

  constructor (
    private readonly vectorIndex: VectorizeIndex,
    private readonly ai: Ai,
  ) {

    if (!vectorIndex) {
      throw new Error(
        'Missing Vectorize binding.')
    }

    if (!ai) {
      throw new Error('Missing AI binding.')
    }
  }

  /**
   * Upserts the given locations into the vector index.
   *
   * @param locations
   */
  public async upsertLocations (locations: Location[]): Promise<number> {

    if (locations.length === 0) {
      return 0
    }

    const AI_BATCH_SIZE = 10
    let successfulUpserts = 0

    for (let i = 0; i < locations.length; i += AI_BATCH_SIZE) {
      const batch = locations.slice(i, i + AI_BATCH_SIZE)

      try {
        const textsToEmbed = batch.map((location) =>
          this.buildTextForEmbedding(location),
        )

        const vectors = await this.embedTextBatch(textsToEmbed)

        const vectorObjects = batch.map((location, idx) => {
          const vector = vectors[idx]
          if (!vector) return null
          return {
            id: location.id,
            values: vector,
            metadata: {
              locationId: location.id,
            },
          }
        }).filter((obj): obj is NonNullable<typeof obj> => obj !== null)

        if (vectorObjects.length > 0) {
          const result = await this.vectorIndex.upsert(vectorObjects)
          console.log('[Repo] Upsert result:', JSON.stringify(result))
          successfulUpserts += vectorObjects.length
        }
      } catch (error) {
        console.error(
          `[CloudflareVectorAIRepository] Failed to process batch starting at index ${i}`,
          error,
        )
      }
    }
    return successfulUpserts
  }

  /**
   * Embeds a batch of texts using the BGE-M3 model.
   *
   * @param texts
   * @private
   */
  private async embedTextBatch (texts: string[]): Promise<number[][]> {
    const response = (await this.ai.run(
      '@cf/baai/bge-m3', {
        text: texts,
        response_format: 'embedding_vector',
      },
    )) as BGEEmbeddingResponse

    const vectors = response.data ?? response.response

    if (!vectors) throw new Error('BGE-M3 invalid response')

    return vectors
  }

  /**
   * Builds the text representation for embedding from a Location.
   *
   * @param location
   * @private
   */
  private buildTextForEmbedding (location: Location): string {
    return `${location.name} | ${location.postcodes.join(', ')}`
  }

}

The repository requires Cloudflare’s Vector Index and AI bindings in its constructor. We will resolve these from Cloudflare’s environment in the queue handler later.

The upsert locations method takes a list of locations from our database, builds text embeddings for each location in batches of 10, maps them to vector objects with the index metadata, and then upserts the vector objects to the vector store.

I’m intentionally excluding unit and integration tests from this demo application for brevity. Before we can test this repository end-to-end, we need to set up our Cloudflare Queue.

How to dispatch and handle Cloudflare Queues in Nuxt

Queues allow us to process tasks in the background, which is essential for decoupling long-running operations from user requests. Like most queue services, Cloudflare Queues includes features such as batching and retries.

In our Nuxt application, we will implement this in two parts:

The Producer (Nitro Task): We will dispatch messages containing just the Location ID from our database. To be efficient, we will use sendBatch to dispatch up to ten IDs at a time.
The Consumer (Nitro Plugin): We will listen for these messages within the same Nuxt application (running as a Cloudflare Worker) using a Nitro plugin to process the data.

Create a Cloudflare Queue and bind the producer and consumer to the same Nuxt application

⚠️ Cloudflare Queues are only available on paid Worker Plans. Currently at $5 per month, plus extra usage charges.

We can create the vector sync queue using Wrangler CLI:

npx wrangler queues create vector-sync-queue

And add the bindings in /wrangler.toml similar to:

[[queues.producers]]
queue = "vector-sync-queue"
binding = "VECTOR_SYNC_QUEUE"

[[queues.consumers]]
queue = "vector-sync-queue"
max_batch_size = 10    # Recommended: Process 10 vectors at a time
max_batch_timeout = 5  # Wait up to 5s to fill a batch

Once you build and deploy, Cloudflare automatically configures your Nuxt Worker to act as the consumer for this queue.

Dispatch to a Cloudflare Queue from a Nuxt application through a Nitro Task

Next, we will write a Nitro Task that dispatches messages to our newly created queue. If you’re not following along from the previous article, make sure you add Nitro’s experimental tasks and async context to Nuxt’s config:

 nitro: {

    preset: 'cloudflare_module',

    experimental: {
      tasks: true,
      asyncContext: true,
    },
  },

Let’s first add some types in /server/types/queues.ts:

import type {
  Ai,
  Queue,
  VectorizeIndex,
  D1Database,
  ExecutionContext,
} from '@cloudflare/workers-types'

export interface CloudflareEnv {
  AI: Ai;
  VECTORIZE: VectorizeIndex;
  VECTOR_SYNC_QUEUE: Queue;
  DB: D1Database

  [key: string]: unknown;
}

export interface VectorSyncQueueMessageBody {
  locationId: string;
}

export interface CloudflareTaskContext {
  cloudflare: {
    env: CloudflareEnv;
    context: ExecutionContext;
  };
}

And create /server/tasks/build-location-embeddings.ts with:

import { drizzle } from 'drizzle-orm/d1'
import * as schema from '~~/server/database/schema'
import type { D1Database, Queue } from '@cloudflare/workers-types'
import { CloudflareTaskContext } from '~~/server/types/queues'

export default defineTask({
  meta: {
    name: 'build-location-embeddings',
    description: 'Queues all location IDs for vector embedding',
  },
  async run(event) {
    const context = event.context as CloudflareTaskContext
    const env = context.cloudflare?.env

    if (!env?.DB) {
      return { error: 'DB binding not found.' }
    }
    const db = drizzle(env.DB as D1Database, { schema })

    // Queue Init
    const queue = env.VECTOR_SYNC_QUEUE as Queue
    if (!queue) {
      return { error: 'Queue binding (VECTOR_SYNC_QUEUE) not found.' }
    }

    try {
      const allLocations = await db.query.locations.findMany({
        columns: { id: true },
      })

      if (allLocations.length === 0) return { result: 'No locations found in DB.' }

      const total = allLocations.length
      console.log(`[Task] Found ${total} locations. Dispatching to queue...`)

      const messages = allLocations.map((location) => ({
        body: { locationId: location.id },
      }))

      const CHUNK_SIZE = 10
      for (let i = 0; i < messages.length; i += CHUNK_SIZE) {
        const batch = messages.slice(i, i + CHUNK_SIZE)
        await queue.sendBatch(batch)
      }

      return {
        result: `Dispatched ${total} locations to VECTOR_SYNC_QUEUE.`,
      }
    } catch (error: any) {
      console.error('[Task] Error during queue dispatch:', error)
      return { error: error.message || 'Unknown error occurred.' }
    }
  },
})

Now that the dispatcher is done. We can trigger the Nitro task from CLI, DevTools, CRON, or programmatically. Since Cloudflare’s Workers AI runs on remote infrastructure, we will build an API endpoint that triggers the Nitro task programmatically after we implement the listener and handler.

💡There are several ways you can safely test Cloudflare’s AI without touching the production environment. You can create bindings specific to a staging environment and also use the getPlatformProxy() wrangler function in tests.

Listen and handle Cloudflare Queue messages from a Nuxt application using a Nitro Plugin

With the queue dispatcher completed, we now need a Nitro Plugin to listen to the dispatched queue and handle it. In the next part of this article series, we will use another queue to synchronise the Agents’ listings. To avoid a massive queue handler class, we will split the listener and handler code.

The following is the handler that depends on the LocationVectorRepository interface we declared earlier.

Create the handler in /server/utils/queueHandlers/VectorSyncQueueHandler.ts with:

import {
  LocationVectorRepository,
} from '~~/server/utils/queueHandlers/repositories/LocationVectorRepository'
import { Location } from '~~/server/database/types'
import * as schema from '~~/server/database/schema'
import { drizzle } from 'drizzle-orm/d1'
import { inArray } from 'drizzle-orm'
import type { D1Database, MessageBatch } from '@cloudflare/workers-types'
import type {
  VectorSyncQueueMessageBody,
  CloudflareEnv,
} from '~~/server/types/queues'

export class VectorSyncQueueHandler {

  constructor (
    private readonly locationVectorRepository: LocationVectorRepository,
  ) {}

  public async handle (
    batch: MessageBatch<VectorSyncQueueMessageBody>,
    env: CloudflareEnv,
  ): Promise<void> {

    const messages = batch.messages
    const db = drizzle(env.DB as D1Database, { schema })

    // Extract Unique IDs
    const locationIds = new Set<string>()
    for (const message of messages) {
      if (message.body?.locationId) {
        locationIds.add(message.body.locationId)
      }
    }

    if (locationIds.size === 0) {
      console.log('[Vector Sync Handler] No valid locationIds found in batch.')
      return
    }

    try {
      const ids = Array.from(locationIds)

      // Fetch Data
      const locationsToUpsert: Location[] = await db.query.locations.findMany({
        where: inArray(schema.locations.id, ids),
      })

      if (locationsToUpsert.length > 0) {
        // Update Vector Index
        await this.locationVectorRepository.upsertLocations(locationsToUpsert)
        console.log(
          `[Vector Sync Handler] Successfully processed ${locationsToUpsert.length} locations.`)
      }

    } catch (error: any) {
      console.error('[Vector Sync Handler] Error:', error.message)
      // Throwing error triggers Cloudflare's automatic queue retry logic
      throw error
    }
  }
}

The handler class above loops through the queue messages, retrieves the full location data from our D1 database, and uses the LocationVectorRepository to upsert the data into the Cloudflare Vectorize store.

Finally, we need a Nitro Plugin to hook into and thus listen to Cloudflare Queues. The plugin will be the entry point for all dispatched Cloudflare Queues and will delegate the specific handler based on the queue name.

Create the plugin at /server/plugins/queue-handler.ts with:

import { VectorSyncQueueHandler } from '~~/server/utils/queueHandlers/VectorSyncQueueHandler'
import { CloudflareVectorAIRepository } from '~~/server/repositories/cloudflare/CloudflareVectorAIRepository'
import type { VectorSyncQueueMessageBody, CloudflareEnv } from '~~/server/types/queues'
import type { MessageBatch } from '@cloudflare/workers-types'

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('cloudflare:queue', async (payload) => {
    const batch = payload.batch
    const env = payload.env as CloudflareEnv

    if (batch.queue === 'vector-sync-queue') {
      console.log(`[Queue] Received batch of ${batch.messages.length} for ${batch.queue}`)

      try {
        const repo = new CloudflareVectorAIRepository(env.VECTORIZE, env.AI)
        const handler = new VectorSyncQueueHandler(repo)

        await handler.handle(
          batch as MessageBatch<VectorSyncQueueMessageBody>,
          env
        )

      } catch (error: any) {
        console.error('[Queue] Error processing batch:', error)
        // This will trigger a retry as per the wrangler.toml queue settings
        throw error
      }
    }
  })
})

Populating the Cloudflare Vectorize index from our Nuxt application

We’ve built everything we need to populate the vector store with our location data, and we’re now ready for the most exciting part of this article.

We will use an API endpoint to run it remotely. Create /server/api/internals/tasks/build-location-embeddings.get.ts with:

export default defineEventHandler(async (event) => {

  const config = useRuntimeConfig();

  if (getHeader(event, 'x-secret') !== config.internalApiSecret) {
    throw createError({ statusCode: 401, statusMessage: 'Unauthorized' });
  }

  const result = await runTask(
    'build-location-embeddings',
    {
      payload: {},
      context: {
        cloudflare: event.context.cloudflare
      }
    }
  );

  return {
    status: 'Build Location Embeddings Task Triggered',
    result,
  };

})

This endpoint triggers our Nitro Task programatically using the runTask() function. Notice we’re using the internal API secret we added as an environment variable to the Cloudflare Worker in the previous article. Also good to note that we’re setting the context to use Cloudflare’s context, since the Nitro Task depends on Cloudflare’s D1 database and the Vector Sync Queue.

With the API endpoint added, go ahead, deploy and test:

pnpm run build
npx wrangler deploy

curl -H "x-secret: [YOUR_SECRET]" "https://[YOUR_WORKER_URL]/api/internals/tasks/build-location-embeddings"

In your Cloudflare worker dashboard, you can see the endpoint trigger and then the queue. We can also verify that the vector index is populated through Wrangler, but you need to wait a couple of minutes for the queue to be processed.

npx wrangler vectorize info locations-index

# Should return something like:

 ⛅️ wrangler 4.53.0
───────────────────
📋 Fetching index info...
┌────────────┬─────────────┬──────────────────────────────────────┬──────────────────────────┐
│ dimensions │ vectorCount │ processedUpToMutation                │ processedUpToDatetime    │
├────────────┼─────────────┼──────────────────────────────────────┼──────────────────────────┤
│ 1024       │ 4           │ aa21b6a6-e25f-4b1c-afa8-98846e801ec6 │ 2025-12-08T09:47:59.146Z │
└────────────┴─────────────┴──────────────────────────────────────┴──────────────────────────┘

npx wrangler vectorize list-vectors locations-index

# Should list the Location ID:

 ⛅️ wrangler 4.53.0
───────────────────
📋 Listing vectors in index 'locations-index'...
┌───┬──────────────────────────────────────┐
│ # │ Vector ID                            │
├───┼──────────────────────────────────────┤
│ 1 │ 1e212fe9-3994-4c7a-a90f-22df3fdee8d5 │
├───┼──────────────────────────────────────┤
│ 2 │ 0168e16d-0571-466d-809b-d717e58d7cab │
├───┼──────────────────────────────────────┤
│ 3 │ 1818d8a6-baf1-4780-b6d7-a8e6f45b3369 │
├───┼──────────────────────────────────────┤
│ 4 │ 76d29d74-5dfe-4a46-b15d-802e976d5250 │
└───┴──────────────────────────────────────┘

Showing 4 of 4 total vectors

Moving forward with implementing our Semantic Matching feature

The final part of this Nuxt & Cloudflare AI Vector Pipeline Series is where the magic happens: we will implement the location semantic matching to tie everything together.

I hope you’re enjoying this series! As always, feel free to ask questions in the comments below or on social media. Please do share this with anyone you think might find it useful, and subscribe if you’d like me to email you when I prepare more code examples and articles.

Peace ✌🏽

Need help with Nuxt & Cloudflare?

Keith Mifsud is an Official Nuxt Agency Partner, learn more.

Nuxt & Cloudflare Vectorize: Setting up D1, Drizzle, and Workers AI

Keith Mifsud — Mon, 29 Dec 2025 13:58:00 +0000

Recently, I started working on a project that requires several data normalisation features. One of which is getting external data and intelligently linking it with existing internal data. Data normalisation is a common requirement, especially recently, since we can augment searches and matches as we deem fit by vectorising our data and querying the vectors with embedding models.

The project is built with Nuxt and hosted on Cloudflare for Edge availability. I initially set it up through NuxtHub Admin for CI/CD and used two core module server composables: hubVectorize() and hubAi(). NuxtHub is sunsetting the NuxtHub Admin at the end of this month. The core module no longer features Cloudflare’s automatic binding setup. Here, I share my experience building a demo application with similar AI requirements using Nuxt on Cloudflare, without the NuxtHub wrapper.

This article is the first of the Nuxt & Cloudflare AI Vector Pipeline Series, a three-part series.

Part one - Nuxt & Cloudflare Vectorize: Setting up D1, Drizzle, and Workers AI
Part two - Nuxt & Cloudflare Queues: Building a Data Sync Pipeline using Vectorize
Part three - Implementing semantic matching in Nuxt with Cloudflare Vectorize

Let’s get going 😁

Building an AI-powered real estate listing Nuxt application

Together, we will build the backend of a fictitious real estate property listings aggregator website. If it were real, it would feature property listings from multiple real estate agencies. The app needs to regularly retrieve a feed of listings from these agencies and add them to the internal properties database. We need to link each agent property to a standardised location.

The challenge with such scenarios is that the data feed may not be categorised as needed; for example, an agent may include the location using a postcode, some might omit the postcode and use the area names, and others might mention the location in the property listing title. In all cases, human errors can prevail, such as typos and incomplete words. We will use AI to match agents' property locations against our locations database to correctly categorise the properties internally.

We will do this on a single Nuxt app, working with a Cloudflare AI worker, a Vector store, a D1 database, and a couple of Queue workers. Almost a monolith, but not precisely so. Since we are binding to multiple Cloudflare services, we can call this a distributed setup 😅.

The messy data challenge

For example, a single property in Shoreditch (an area of London) might appear in three completely different ways across our incoming feeds:

Agent A: Uses the postcode "E1 6AN" as a specific postcode attribute.
Agent B: Uses the standard area name "Shoreditch" as a specific location attribute.
Agent C: Does not include postcode or location but instead adds the location as part of the property title “Hidden Gem 💎 (around Shoreditch)”.

While our database schema includes a list of postcodes mapped to areas, it lacks a native way to categorise a property's locations from a mix of postcodes, addresses, and descriptions, especially when they conflict or are misspelt. Without a smart normalisation layer, we cannot reliably categorise these listings, leaving our database fragmented and preventing us from maintaining a clean "Source of Truth" for our application.

💡 It is not always ideal to process business logic on an Edge provider such as Cloudflare, and it is often unnecessary. Let me know in the comments or on socials if you’d like me to write about some of the many alternative architectures that preserve Edge availability while reducing development and operational infrastructure costs.

In this article, we’ll set up the internal database, schema migrations, and ORM, deploy it to a Cloudflare D1 database, and prepare the AI worker binding for later.

The code for the entire demo application is publicly available on GitHub: Nuxt & Cloudflare AI Vector Pipeline Series - give it a ⭐.

Drizzle DB Schema on Cloudflare D1

Let’s start with a fresh Nuxt install, install Drizzle, Drizzle Kit for generating database schema migration files, and the Cloudflare / Wrangler dependencies:

pnpm dlx nuxi@latest init nuxt-data-ai-sync
cd nuxt-data-ai-sync
pnpm add drizzle-orm postgres
pnpm add -D drizzle-kit wrangler @types/node nitro-cloudflare-dev

We can also configure Nuxt to use Nitro’s experimental features, which we will need later. Mainly tasks and asyncContext. Your /nuxt.config.ts should include the correct Nitro preset and Nitro’s experimental features:

export default defineNuxtConfig({
  compatibilityDate: '2025-12-06',
  devtools: { enabled: true },
  modules: [
    'nitro-cloudflare-dev',
  ],
  nitro: {

    preset: 'cloudflare_module',

    experimental: {
      tasks: true,
      asyncContext: true,
    },
  },
})

Deploy Nuxt as a worker on Cloudflare

Now’s a good time to add the initial Wrangler configuration so we can deploy to Cloudflare for the first time. Create the /wrangler.toml file at the project’s root and fill in the initial content:

name = "property-sync-app"
main = "./.output/server/index.mjs"
compatibility_date = "2025-08-02"
account_id = "xxxx" # <-- your account ID here

assets = { directory = "./.output/public", binding = "ASSETS" }

Now we can build and deploy our so far useless app on Cloudflare:

pnpm build
npx wrangler deploy

The output should look similar to the following;

⛅️ wrangler 4.53.0
───────────────────
🌀 Building list of assets...
✨ Read 14 files from the assets directory /home/keithmifsud/projects/keithmifsud/nuxt-data-ai-sync/.output/public
🌀 Starting asset upload...
🌀 Found 10 new or modified static assets to upload. Proceeding with upload...
+ /robots.txt
+ /_nuxt/error-500.pPqbDRtt.css
+ /favicon.ico
+ /_nuxt/b-_NRaZA.js
+ /_nuxt/builds/latest.json
+ /_nuxt/error-404.ClksgLvO.css
+ /_nuxt/entry.ClxZN-oh.css
+ /_nuxt/builds/meta/23e8b19f-c1c2-45ca-83bc-9cad6a8adfcf.json
+ /_nuxt/Cv0oBIJc.js
+ /_nuxt/BVjrWf75.js
Uploaded 3 of 10 assets
Uploaded 6 of 10 assets
Uploaded 10 of 10 assets
✨ Success! Uploaded 10 files (3.21 sec)

Total Upload: 685.31 KiB / gzip: 184.57 KiB
Worker Startup Time: 7 ms
Your Worker has access to the following bindings:
Binding            Resource      
env.ASSETS         Assets        

Uploaded property-sync-app (15.49 sec)
Deployed property-sync-app triggers (4.65 sec)
  https://property-sync-app.mifsud-k.workers.dev
Current Version ID: 3d88b77c-2d1f-4c8e-8f01-22823a4a1c61

Thus, you can test that the URL shows the Nuxt welcome page.

I like to get the noticeable housekeeping tasks tackled as soon as possible. Since we will definitely be dealing with queued processes, we will be working across different contexts (Cloudflare Workers using the V8 Engine and Nitro Runtime using Node.js) asynchronously. Give the worker settings in Cloudflare’s UI a once-over and make sure your runtime has the nodejs_compat flag enabled.

Modelling the database schema using Drizzle ORM

Since our schema is relatively small, we can place it in a single file. Create /server/database/schema.ts with:

import { sqliteTable, text, integer } from 'drizzle-orm/sqlite-core'
import { relations, sql } from 'drizzle-orm'

const generateUUID = () => crypto.randomUUID()

// Agents
export const agents = sqliteTable('agents', {
  id: text('id').primaryKey().$defaultFn(generateUUID),
  name: text('name').notNull(),
  apiRoute: text('api_route').notNull().unique(),
  createdAt: text('created_at').default(sql`(CURRENT_TIMESTAMP)`),
})

// Locations (Canonical)
export const locations = sqliteTable('locations', {
  id: text('id').primaryKey().$defaultFn(generateUUID),
  name: text('name').notNull(),
  // D1/SQLite stores arrays as strings. We handle parsing in the app.
  postcodes: text('postcodes', { mode: 'json' }).$type<string[]>().notNull(),
  createdAt: text('created_at').default(sql`(CURRENT_TIMESTAMP)`),
})

// Properties (Incoming)
export const properties = sqliteTable('properties', {
  id: text('id').primaryKey().$defaultFn(generateUUID),
  externalRef: text('external_ref').notNull(), // Agent's ID for the property
  agentId: text('agent_id').references(() => agents.id).notNull(),
  locationId: text('location_id').references(() => locations.id).notNull(),
  title: text('title').notNull(),
  originalLocation: text('original_location').notNull(), // The messy string
  createdAt: text('created_at').default(sql`(CURRENT_TIMESTAMP)`),
})

export const agentsRelations = relations(agents, ({ many }) => ({
  properties: many(properties),
}))

export const propertiesRelations = relations(properties, ({ one }) => ({
  agent: one(agents, {
    fields: [properties.agentId],
    references: [agents.id],
  }),
  location: one(locations, {
    fields: [properties.locationId],
    references: [locations.id],
  }),
}))

Note that SQLite does not support UUID types, which is why we’re using the text type and a function to generate the UUIDs.

We will populate the agents' table with a list of agents from whom we'll fetch a listings feed. The locations table will hold our known locations, which we'll later use to populate the vector index and eventually match against the agent property listings. Finally, the properties table will hold agent properties with a resolved location.

I’ve also added a couple of soft relations as an example.

While we're here, we can generate the database migration files using Drizzle Kit. Generating them is relatively straightforward (unless you're working in a team; Drizzle Kit can result in several conflicts with the migration files order 😮‍💨). Just run npx drizzle-kit generate to generate the migration files from the schema we just wrote.

Running Drizzle migrations on a Cloudflare D1 database

We can now create a new D1 database on Cloudflare. We can do this using Wrangler CLI:

npx wrangler d1 create property-sync-db

And add the database binding to /wrangler.toml:

[[d1_databases]]
binding = "DB"
database_name = "property-sync-db"
database_id = "the-id-returned-from-the-wrangler-command"
migrations_dir = "server/database/migrations"

Creating the D1 database on Cloudflare does not automatically bind it to our worker. Bindings are added on deployment based on the wrangler.toml file entries. Notice that we're also specifying the migrations directory location, which we will need soon.

⚠️ Note: Cloudflare’s D1 is not always a good fit. Scaling large datasets requires manual sharding, and as with all SQLite databases, D1 lacks several key features that production-grade applications often need.

We also need to enter some database credentials in /drizzle.config.ts.

import { defineConfig } from 'drizzle-kit'

export default defineConfig({
  dialect: 'sqlite',
  schema: './server/database/schema.ts',
  out: './server/database/migrations',
  driver: 'd1-http',

  dbCredentials: {
      accountId: process.env.CLOUDFLARE_ACCOUNT_ID!,
      databaseId: process.env.CLOUDFLARE_DATABASE_ID!,
      token: process.env.DRIZZLE_API_TOKEN!,
  },
})

You should add the environment variables and use them instead of adding them directly in the config file. This so you can swap them for different environments. We don’t have dbCredentials.token, but migrations seem to fail if not included. My .env file looks like this:

CLOUDFLARE_ACCOUNT_ID="my_cloudflare_account_id"
CLOUDFLARE_DATABASE_ID="the_database_id_from_cloudflare"
# Notice I don't have DRIZZLE_API_TOKEN

And with that, we can run the database migrations directly on the D1 instance by running:

npx wrangler d1 migrations apply property-sync-db --remote

Let’s seed some data on our Cloudflare D1 Database

I'm going to create an API endpoint to seed some needed data into the D1 database. You'd obviously not do this, this way, in a real application. Still, this way we can move this task forward faster and lay some groundwork for when we need other internal API endpoints later, since Cloudflare's AI features don't run in dev environments. However, to make it harder for attackers, we should add an environment variable to the Cloudflare worker.

We can add the secret using Wrangler:

npx wrangler secret put NUXT_INTERNAL_API_SECRET

Or from Cloudflare’s dashboard, navigate to Worker & Pages, select this worker, and go to settings. Then add an encrypted environment variable named NUXT_INTERNAL_API_SECRET with a secret string. Feel free to add the same variable locally if you intend to use the upcoming API endpoint from localhost.

We need to seed some dummy agents and a few locations so we can generate text embeddings, add the locations to the vector store, and later use AI to resolve the messy listings from the agents' feeds.

Create an API endpoint at: /server/api/internals/seed.get.ts and add the following contents:

import { drizzle } from 'drizzle-orm/d1'
import { agents, locations } from '~~/server/database/schema'

export default defineEventHandler(async (event) => {

  const config = useRuntimeConfig()

  if (getHeader(event, 'x-secret') !== config.internalApiSecret) {
    throw createError({ statusCode: 401, statusMessage: 'Unauthorized' })
  }

  const env = event.context.cloudflare.env

  if (!env.DB) {
    throw createError(
      { statusCode: 500, statusMessage: 'Database not found in environment' })
  }

  const db = drizzle(env.DB)

  console.log('🌱 Starting Seed...')

  // 3. Seed Agents
  const agentsData = [
    {
      name: 'The Formalist',
      apiRoute: '/api/agents/formalist',
    },
    {
      name: 'The Traditionalist',
      apiRoute: '/api/agents/traditionalist',
    },
    {
      name: 'The Marketer',
      apiRoute: '/api/agents/marketer',
    },
  ]
  await db.insert(agents).values(agentsData)

  // 4. Seed Locations (Corrected Data)
  const locationsData = [
    {
      name: 'Shoreditch',
      postcodes: ['E1', 'E2', 'EC1', 'EC2', 'N1'],
    },
    {
      name: 'Canary Wharf',
      postcodes: ['E14'],
    },
    {
      name: 'Brixton',
      postcodes: ['SW2', 'SW9'],
    },
    {
      name: 'Notting Hill',
      postcodes: ['W10', 'W11', 'W2'],
    },
  ]
  await db.insert(locations).values(locationsData)

  return {
    success: true,
    message: 'Database Seeded',
    stats: {
      agents: agentsData.length,
      locations: locationsData.length,
    },
  }
})

That’s it, build, deploy, and run the seeder:

pnpm build
npx wrangler deploy

curl -H "x-secret: [YOUR_SECRET]" "https://[YOUR_WORKER_URL]/api/internals/seed

You can now see the seeded data through Cloudflare’s Database Explorer by navigating to:

Storage & databases → D1 SQL database → property-sync-db → Explore Data

Cloudflare Workers AI

Although this is not a tutorial about AI, it is important to note the difference between Cloudflare Workers AI and Cloudflare Vectorize. Vector stores are "simply" a database of vectors; Pinecone{target="_blank"} is a popular alternative to Cloudflare Vectorize. In contrast, Workers AI is the compute part, the processor that generates text embedding from our data, which can later be stored as vectors.

In short, Vectorize is the memory, and Workers AI is the brain. With standalone solutions like Pinecone, you generally still need separate services for the memory and the brain (e.g., Pinecone + OpenAI). The Cloudflare ecosystem bundles the generation of embeddings (Workers AI) and their storage (Vectorize) into a single, low-latency workflow.

Since Workers AI is a built-in platform service, we only need to specify the binding in our wrangler.toml config to access it directly form our Nuxt application.

Go ahead and add the binding in wrangler.toml:

[ai]
binding = "AI"

This binding will be effective on our next deployment.

Moving forward with storing vectors on Cloudflare’s Vectorize Index

In the next part of this Nuxt & Cloudflare AI Vector Pipeline Series, we will be building a Background Data Sync Pipeline using Vectorize. In the article, we will also take a deep dive into working with Cloudflare Queues from our Nuxt application and, most importantly, listen and handle the dispatched Queue messages from the same Nuxt application, i.e., the same Cloudflare Worker.

I hope you found this article helpful. Feel free to ask me anything in the comments below or on Social. Please share it with your peers and subscribe to get notified when I publish new articles 💙.

Need help with Nuxt & Cloudflare?

Keith Mifsud is an Official Nuxt Agency Partner, learn more.

Build and deploy a static online shop with Nuxt3 using Pinia Store and Stripe Checkout to Firebase

Keith Mifsud — Sun, 20 Nov 2022 05:12:00 +0000

Nuxt3 final release is due in a few days and is already released when you're reading this. I've been looking forward to Nuxt3 for almost a year as I've been experimenting with it since the alpha release. I also tried to use Nuxt3 beta release on a couple of projects earlier this year but, unfortunately I was faced with several issues mainly related to missing plugins such as GraphQL tools and issues with static site generation.

The RC versions have started coming in the past few months, and so I decided to try out the static site generation feature on Nuxt3 by building a small shop, accept Stripe payments and deploy it on Firebase.

I found the process of building the online shop with Nuxt3 and deploying the statically generated pages to Firebase to be a breeze. I will be sharing the steps I've taken to create this website here, with you all. A useful point of reference for all new Nuxt3 static websites.

🥱 TLDR

All the code for setting up a static website with Stripe Checkout using NuxtJs is available on GitHub.

Getting started - installing Nuxt3, Pinia Store and TailwindCSS

Installing Nuxt3 is, as expected, straight forward.

The official installation docs require Node.js version 16.11 or above. If like myself you have several projects requiring different Node.js versions, you can consider using NVM (Node Version Manage) to easily switch between Node.js versions.

npx nuxi init my-shop-name

cd my-shop-name

npm install

npm run dev

The new Nuxt3 site will be available at http://localhost:3000.

Installing and configuring a Pinia Store on Nuxt3

This simple static online shop will not be fetching the products or any content from an external source, instead I have all the data in a json file, however, I still wanted to use Pinia Store for a couple of reasons. The first being I wanted to see if it is working as expected on Nuxt3 as I use it on almost on every VueJS project. Second, I think it is easier to swap the data sources from within stores' actions instead of the components directly.

Just run:

npm install @pinia/nuxt

to install the plugin and then add it as a module (with auto-import) to the Nuxt config file nuxt.config.ts.

export default defineNuxtConfig({
  modules: [
    [
      '@pinia/nuxt',
      {
        autoImports: [
          ['defineStore', 'definePiniaStore'],
        ],
      },
    ],
  ],
})

I then added the initial products store in ./stores/products.js.

export const useProductsStore = definePiniaStore(
  'products-store',
  {
    state : () => (
      {
        products: [],
      }
    ),
    getters: {},
    actions: {},
  }
)

Planning the technology stack for an Event Sourcing project

Keith Mifsud — Fri, 13 Sep 2019 05:48:38 +0000

The most common question I encounter when training or consulting with developers, Engineers and Software development laboratories about a new Event Sourcing project is how and where do we start. This question makes so much sense. I remember trying to get my head around Object Oriented Programming in practice (not the crap I learnt at school), let alone understanding Domain-Driven Design!

Almost a decade ago, I was building a relatively complex online and offline sales and logistics system for what was, back in the day one of my largest clients: a manufacturer, wholesaler and distributor of industrial Refrigeration and Air-conditioning systems. Three months in the project, I had my team set up with three Software Engineers, two backend web developers and one frontend developer. We were making excellent progress. We were building this system using a DDD model and Laravel as the application framework. It was good, and we were productive. Then, the client hired an internal CTO. He asked to meet me, and I obliged. He asked me how it was working in Event Sourcing with PHP. My answer was more of a series of questions 😳.

I already knew about Event Sourcing, I read the books but I never, ever needed to implement such a thing. I did not know where to start. I mean, I had no idea. I knew the concept and the theory, but that was it.

In my previous article, I explained how I get started with a new Event Sourcing Project. With some theory and Event Storming covered, it is time to start thinking about other equally important factors, the Technology Stack for an Event Sourcing project.

I chose PHP as the primary backend programming language. However, I am confident that the planning factors apply to any programming language you'd like to use for your Event Sourcing project. In this article, I will outline what the considerations for planning the Infrastructure and the Application glue in between are. I will also describe and provide a working example of my local environment deployment, code structure, and what I deem as the starting point for such a system, the testing framework.

How to plan the infrastructure services for an event sourcing project
Arguably, I believe that in the majority of cases, when working with a team, preparing for these requirements will depend on several factors and other project stakeholder's needs. Although I am making these decisions by myself for this project, the process is similar for teams and projects commissioned by your clients. The main difference is that I can choose which backend services I am most comfortable using and that I don't have external dependencies beyond my control. Such as migrating or integrating data from an MSSQL database into a MySQL Database 😡.

During my career, I had an equal number of opportunities for developing new systems and migrating legacy systems into an event sourcing solution. In my opinion, the latter is a lot more complicated. You can get in touch with me if you need help with refactoring a legacy system to use an event sourcing pattern and also to migrate your existing data.
I intentionally kept this example Domain simple so to not distract you from the primary topic, which is to build an Event Sourcing project with practical examples. The example code in this project is a sample from a commercial project I manage. Please feel free to ask me questions in the comments section further below about this or other Business Domains. With this out of the way, lets us start planning the Technology Stack for our example project because we have a lot to cover.

Event store database
The most critical persistent storage for Event Sourcing is the Event Store Database. As I'm sure you are aware, there are several different database engines we can use for running our event store. I will not be comparing any of these engines in this post because all we need to know is that we need an append optimised database.

Given the immutable nature of Event Sourcing, we certainly don't need to update any of our persisted events. We also don't need to run any complex read queries. As I will mention in a future article, the Event Store implementation will query the persisted events as identified by the Aggregate's UUID (Universally Unique Identifier). That is quite a simple query.

Start a new project with an Event Sourcing Architecture

Keith Mifsud — Wed, 17 Jul 2019 05:23:25 +0000

Event Sourcing example project

The best way I can help you, the Engineer, the Product Owner or as I will be throughout this series of articles, both, is with an example event sourcing project. There are several components required in building an event sourcing project and the best way for you to be able to plan and estimate your project is with understanding the amount of work required to build and adapt these components.

To illustrate what it takes to plan and build an event-sourced application, I chose to use a simple To-Do application... I'm kidding 😂, To-Do applications are all over the web and they are too basic to justify this pattern. Instead, I, together with you, will be building a Classified Listing web-based application. A scaled-down version of GumTree .
With this example project, I will focus on the main business problem, the Listing Domain. Throughout a series of articles, I will show you the process I will follow to plan, design and develop an event-sourced Listing application. I intentionally chose this domain (or problem) because I know it as much as you do, unless, of course, you created GumTree yourself 😂.

In this series of articles, I will explain everything needed to have a Listing Web API, providing Members with the ability, of drafting and publishing a Listing, to advertise an item they wish to sell or purchase. Therefore, I will write about planning, how to solve the problems (or provide the opportunities) for this business, setting up a development environment, developing a Common/Shared Model, developing the first Aggregate, applying and dispatching events using a CQRS approach implementing concrete implementations for the system's requirements, testing, cross-domain consequences, read models and the access through the application.

I should get on with it, right? But where to start? From a business and service offering perspective, I would start with Event Storming. However, implementing the infrastructure and Integrations with external services should also be planned to ensure the feasibility of the project.

We are in luck as our Domain and system do not make use of external services. However, I want to highlight that this is a rare scenario! Depending on the product you are building, you may experience integration issues. Just recently, I was working on a Smart Condominium system whose internal hardware provider did not have a public API. The point is that, yes, most common and general service providers have an API but in the real world, a lot of vendors protect their IP (Intellectual property) by not offering an accessible API.

Whether you're developing an Event Sourcing Domain or an application using any pattern, it pays to determine the integration possibilities upfront.

This example Event Sourcing project does not have any external services' requirements and the Infrastructure requirements will be easy to implement because they are quite common.