feat: add tail worker that incorporates the queue (#14)

* add tail worker that incorporates the queue

* deploy also the tail worker

Author: vahidlazio

confidence-cloudflare-resolver/deployer/script.sh

@@ -309,6 +309,11 @@ RUSTFLAGS='--cfg getrandom_backend="wasm_js"' worker-build --release
 
 # only deploy if NO_DEPLOY is not set
 if test -z "$NO_DEPLOY"; then
+     # deploy the tail worker
+     cd tail
+     wrangler deploy
+     # deploy the resolver worker
+     cd ..
      wrangler deploy
 else
      echo "NO_DEPLOY is set, skipping deploy"

confidence-cloudflare-resolver/src/lib.rs

@@ -214,7 +214,7 @@ pub async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
         && aggregated.client_resolve_info.is_empty())
     {
         let converted = serde_json::to_string(&aggregated)?;
-        FLAGS_LOGS_QUEUE.get().unwrap().send(converted).await?;
+        console_log!("FLAGS_LOGS_QUEUE:{}", converted);
     }
 
     response
@@ -249,7 +249,6 @@ pub async fn consume_flag_logs_queue(
 
     Ok(())
 }
-
 fn get_token(client_id: &str, client_secret: &str) -> String {
     let combined = format!("{}:{}", client_id, client_secret);
     let encoded = STANDARD.encode(combined.as_bytes());
@@ -268,7 +267,6 @@ async fn send_flags_logs(
     client_secret: &str,
     message: WriteFlagLogsRequest,
 ) -> Result<Response> {
-    console_log!("Sending logs {:?}", message);
     let resolve_url = "https://resolver.confidence.dev/v1/flagLogs:write";
     let mut init = RequestInit::new();
     let headers = Headers::new();

confidence-cloudflare-resolver/tail/src/index.js

@@ -0,0 +1,47 @@
+export default {
+    async tail(events, env, ctx) {
+        // Retrieve the queue binding
+        const queue = env.flag_logs_queue;
+        if (!queue) {
+            console.error("LOG_QUEUE binding is missing");
+            return;
+        }
+
+        for (const event of events) {
+            try {
+                // Filter events from confidence-cloudflare-resolver only
+                if (event.scriptName !== "confidence-cloudflare-resolver") {
+                    continue;
+                }
+
+                // Process each log entry in the event
+                for (const logEntry of event.logs || []) {
+                    try {
+                        // Check if the log message starts with FLAGS_LOGS_QUEUE:
+                        const messageStr = Array.isArray(logEntry.message)
+                            ? logEntry.message.join(' ')
+                            : String(logEntry.message || '');
+
+                        if (messageStr.startsWith('FLAGS_LOGS_QUEUE:')) {
+                            // Remove the FLAGS_LOGS_QUEUE: prefix
+                            const cleanedMessage = messageStr.substring('FLAGS_LOGS_QUEUE:'.length).trim();
+
+                            try {
+                                // Parse the JSON payload to validate it's valid JSON
+                                JSON.parse(cleanedMessage);
+                                await queue.send(cleanedMessage);
+                            } catch (parseErr) {
+                                console.error("Failed to parse JSON payload:", parseErr);
+                                console.error("Raw message was:", cleanedMessage);
+                            }
+                        }
+                    } catch (logErr) {
+                        console.error("Failed to process log entry:", logErr);
+                    }
+                }
+            } catch (err) {
+                console.error("Failed to process event:", err);
+            }
+        }
+    }
+};

confidence-cloudflare-resolver/tail/wrangler.toml

@@ -0,0 +1,12 @@
+name = "confidence-resolver-tail"
+type = "javascript"
+compatibility_date = "2025-01-28"
+workers_dev = true
+main = "src/index.js"
+
+[observability]
+enabled = true
+
+[[queues.producers]]
+queue = "flag-logs-queue"
+binding = "flag_logs_queue"

confidence-cloudflare-resolver/wrangler.toml

@@ -1,16 +1,11 @@
 name = "confidence-cloudflare-resolver"
 main = "build/worker/shim.mjs"
 compatibility_date = "2025-01-28"
+tail_consumers = [{service = "confidence-resolver-tail"}]
 
 [observability]
 enabled = true
 
-
-
-[[queues.producers]]
-queue = "flag-logs-queue"
-binding = "flag_logs_queue"
-
 [[queues.consumers]]
 queue = "flag-logs-queue"
 max_batch_size = 100 # max number of messages in a batch