feat: mailbox

Signed-off-by: pyranota <pyra@duck.com>

backend/migrations/20251028105101_mailbox.down.sql

@@ -0,0 +1,4 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS mailbox;
+DROP TYPE IF EXISTS mailbox_type;
+DROP SEQUENCE IF EXISTS mailbox_id_seq;

backend/migrations/20251028105101_mailbox.up.sql

@@ -0,0 +1,18 @@
+-- Add up migration script here
+CREATE SEQUENCE IF NOT EXISTS mailbox_id_seq;
+
+CREATE TYPE mailbox_type AS ENUM (
+    'trigger',
+    'debouncing_stale_data'
+);
+
+CREATE TABLE mailbox(
+    message_id BIGINT DEFAULT nextval('mailbox_id_seq'),
+    mailbox_id TEXT, 
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), -- Indicates position in stack
+    type mailbox_type NOT NULL, -- Type of mailbox
+    payload JSONB NOT NULL, -- Payload of specific message
+    PRIMARY KEY (message_id)
+);
+ 
+

backend/windmill-common/src/lib.rs

@@ -59,6 +59,7 @@ pub mod git_sync_ee;
 pub mod git_sync_oss;
 pub mod jobs;
 pub mod jwt;
+pub mod mailbox;
 pub mod mcp_client;
 pub mod more_serde;
 pub mod oauth2;

backend/windmill-common/src/mailbox.rs

@@ -0,0 +1,59 @@
+use sqlx::Postgres;
+
+use crate::error;
+
+pub struct Mailbox {
+    mailbox_id: Option<String>,
+    mailbox_type: MailboxType,
+}
+
+pub type MsgPayload = serde_json::Value;
+
+#[derive(sqlx::FromRow)]
+pub struct MailboxMsg {
+    payload: MsgPayload,
+    created_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(sqlx::Type, Clone, Copy)]
+#[sqlx(rename_all = "snake_case", type_name = "mailbox_type")]
+pub enum MailboxType {
+    Trigger,
+    DebouncingStaleData,
+}
+
+impl Mailbox {
+    pub fn open(mailbox_id: Option<&str>, mailbox_type: MailboxType) -> Self {
+        Self { mailbox_id: mailbox_id.map(str::to_owned), mailbox_type }
+    }
+    pub async fn push<'c>(
+        &self,
+        payload: MsgPayload,
+        e: impl sqlx::Executor<'c, Database = Postgres>,
+    ) -> error::Result<()> {
+        sqlx::query!(
+            r#"INSERT INTO mailbox(mailbox_id, type, payload) VALUES ($1, $2, $3)"#,
+            self.mailbox_id.as_ref(),
+            self.mailbox_type as MailboxType,
+            payload
+        )
+        .execute(e)
+        .await?;
+
+        Ok(())
+    }
+    pub async fn pull<'c>(
+        &self,
+        e: impl sqlx::Executor<'c, Database = Postgres>,
+    ) -> error::Result<Option<MailboxMsg>> {
+        Ok(sqlx::query_as!(
+            MailboxMsg,
+            r#"DELETE FROM mailbox WHERE type = $1 AND mailbox_id = $2 RETURNING (message_id, created_at)"#,
+            self.mailbox_type as MailboxType,
+            self.mailbox_id.as_ref(),
+        )
+        .fetch_optional(e)
+        .await?)
+    }
+    pub async fn pull_all() {}
+}