feat: event channel and queue tests to use new store structure

- Updated test imports to reflect changes in the event channel and queue structure.
- Replaced instances of `TableEventQueue` with `SyncTableEventQueue` in tests.
- Changed error handling in normalization functions to raise `EventChannelError` instead of `ValueError`.
- Introduced new tests for hook methods in `BaseEventQueueStore` to ensure DDL generation consistency.
- Added tests for dialect-specific overrides in string, integer, and timestamp types for various database adapters.
- Ensured all required columns and default values are correctly included in DDL generation tests.

Author: cofin

AGENTS.md

@@ -970,26 +970,37 @@ Available backends:
 PostgreSQL adapters (asyncpg, psycopg, psqlpy) default to `listen_notify`.
 All other adapters default to `table_queue`.
 
-**Stores** generate adapter-specific DDL for the queue table:
+**Stores** generate adapter-specific DDL using a hook-based pattern:
 
 ```python
 # In sqlspec/adapters/{adapter}/events/store.py
 class AdapterEventQueueStore(BaseEventQueueStore[AdapterConfig]):
     __slots__ = ()
 
+    # REQUIRED: Return (payload_type, metadata_type, timestamp_type)
     def _column_types(self) -> tuple[str, str, str]:
-        # Return (payload_type, metadata_type, timestamp_type)
         return "JSONB", "JSONB", "TIMESTAMPTZ"  # PostgreSQL
 
-    def _build_create_table_sql(self) -> str:
-        # Override for database-specific DDL syntax
-        return super()._build_create_table_sql()
+    # OPTIONAL hooks for dialect variations:
+    def _string_type(self, length: int) -> str:
+        return f"VARCHAR({length})"  # Override for STRING(N), VARCHAR2(N), etc.
 
-    def _wrap_create_statement(self, statement: str, object_type: str) -> str:
-        # Wrap with IF NOT EXISTS, PL/SQL blocks, etc.
-        return statement.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)
+    def _integer_type(self) -> str:
+        return "INTEGER"  # Override for INT64, NUMBER(10), etc.
+
+    def _timestamp_default(self) -> str:
+        return "CURRENT_TIMESTAMP"  # Override for CURRENT_TIMESTAMP(6), SYSTIMESTAMP, etc.
+
+    def _primary_key_syntax(self) -> str:
+        return ""  # Override for " PRIMARY KEY (event_id)" if PK must be inline
+
+    def _table_clause(self) -> str:
+        return ""  # Override for " CLUSTER BY ..." or " INMEMORY ..."
 ```
 
+Most adapters only override `_column_types()`. Complex dialects may override
+`_build_create_table_sql()` directly (Oracle PL/SQL, BigQuery CLUSTER BY, Spanner no-DEFAULT).
+
 **Backend factory pattern** for native backends:
 
 ```python

docs/guides/events/database-event-channels.md

@@ -318,13 +318,23 @@ use `migration_config={"exclude_extensions": ["events"]}`.
 
 ### Event Queue Stores
 
-Stores generate adapter-specific DDL for the queue table. Each adapter has
-a store class in `sqlspec/adapters/{adapter}/events/store.py` that handles:
-
-- Column type mapping (JSON, JSONB, CLOB, etc.)
-- Timestamp types (TIMESTAMPTZ, DATETIME, TIMESTAMP)
-- Index creation strategies
-- Database-specific DDL wrapping (IF NOT EXISTS, PL/SQL blocks, etc.)
+Stores generate adapter-specific DDL for the queue table using a hook-based
+pattern. The base class `BaseEventQueueStore` provides a template for DDL
+generation, and adapters override hook methods for dialect-specific variations:
+
+**Required hook:**
+- `_column_types()` - Return tuple of (payload_type, metadata_type, timestamp_type)
+
+**Optional hooks for dialect variations:**
+- `_string_type(length)` - String type syntax (default: `VARCHAR(N)`)
+- `_integer_type()` - Integer type syntax (default: `INTEGER`)
+- `_timestamp_default()` - Timestamp default expression (default: `CURRENT_TIMESTAMP`)
+- `_primary_key_syntax()` - Inline PRIMARY KEY clause (default: empty, PK on column)
+- `_table_clause()` - Additional table options (default: empty)
+
+Most adapters only need to override `_column_types()`. Complex dialects
+(Oracle PL/SQL, BigQuery CLUSTER BY, Spanner no-DEFAULT) may override
+`_build_create_table_sql()` directly.
 
 Example store implementations:
 

sqlspec/adapters/asyncmy/events/store.py

@@ -55,31 +55,16 @@ def _column_types(self) -> tuple[str, str, str]:
         """
         return "JSON", "JSON", "DATETIME(6)"
 
-    def _build_create_table_sql(self) -> str:
-        """Build MySQL-specific CREATE TABLE SQL.
+    def _timestamp_default(self) -> str:
+        """Return MySQL timestamp default expression.
 
         MySQL requires CURRENT_TIMESTAMP(6) for DATETIME(6) columns,
         not just CURRENT_TIMESTAMP which is only valid for TIMESTAMP type.
 
         Returns:
-            CREATE TABLE SQL statement with MySQL-specific defaults.
+            MySQL-specific timestamp default with microsecond precision.
         """
-        payload_type, metadata_type, timestamp_type = self._column_types()
-        table_clause = self._table_clause()
-        return (
-            f"CREATE TABLE {self.table_name} ("
-            "event_id VARCHAR(64) PRIMARY KEY,"
-            " channel VARCHAR(128) NOT NULL,"
-            f" payload_json {payload_type} NOT NULL,"
-            f" metadata_json {metadata_type},"
-            " status VARCHAR(32) NOT NULL DEFAULT 'pending',"
-            f" available_at {timestamp_type} NOT NULL DEFAULT CURRENT_TIMESTAMP(6),"
-            f" lease_expires_at {timestamp_type},"
-            " attempts INTEGER NOT NULL DEFAULT 0,"
-            f" created_at {timestamp_type} NOT NULL DEFAULT CURRENT_TIMESTAMP(6),"
-            f" acknowledged_at {timestamp_type}"
-            f") {table_clause}"
-        )
+        return "CURRENT_TIMESTAMP(6)"
 
     def _build_index_sql(self) -> str | None:
         """Build MySQL conditional index creation SQL.

sqlspec/adapters/asyncpg/events/backend.py

@@ -10,11 +10,11 @@
 from sqlspec.exceptions import EventChannelError, ImproperConfigurationError
 from sqlspec.extensions.events import EventMessage
 from sqlspec.extensions.events._payload import decode_notify_payload, encode_notify_payload
-from sqlspec.extensions.events._queue import AsyncQueueEventBackend, TableEventQueue, build_queue_backend
+from sqlspec.extensions.events._queue import AsyncTableEventQueue, build_queue_backend
 from sqlspec.extensions.events._store import normalize_event_channel_name
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.serializers import to_json
-from sqlspec.utils.type_guards import has_attr, is_notification
+from sqlspec.utils.type_guards import is_notification
 from sqlspec.utils.uuids import uuid4
 
 if TYPE_CHECKING:
@@ -34,7 +34,7 @@ class AsyncpgHybridEventsBackend:
     supports_async = True
     backend_name = "listen_notify_durable"
 
-    def __init__(self, config: "AsyncpgConfig", queue: "AsyncQueueEventBackend") -> None:
+    def __init__(self, config: "AsyncpgConfig", queue: "AsyncTableEventQueue") -> None:
         if not config.is_async:
             msg = "Asyncpg hybrid backend requires an async adapter"
             raise ImproperConfigurationError(msg)
@@ -56,7 +56,7 @@ async def dequeue(self, channel: str, poll_interval: float) -> EventMessage | No
         if notifies_queue is not None:
             message = await self._dequeue_with_notifies(connection, channel, poll_interval)
         else:
-            message = await self._queue.dequeue(channel, poll_interval=poll_interval)
+            message = await self._queue.dequeue(channel, poll_interval)
         return message
 
     async def ack(self, event_id: str) -> None:
@@ -90,37 +90,27 @@ async def _publish_durable(
     ) -> None:
         """Insert event into durable queue and send NOTIFY wakeup signal."""
         now = datetime.now(timezone.utc)
-        queue_backend = self._get_queue_backend()
-        statement_config = queue_backend.statement_config
         async with self._config.provide_session() as driver:
             await driver.execute(
                 SQL(
-                    queue_backend._upsert_sql,
+                    self._queue._upsert_sql,
                     {
                         "event_id": event_id,
                         "channel": channel,
-                        "payload_json": queue_backend._encode_json(payload),
-                        "metadata_json": queue_backend._encode_json(metadata),
+                        "payload_json": to_json(payload),
+                        "metadata_json": to_json(metadata) if metadata else None,
                         "status": "pending",
                         "available_at": now,
                         "lease_expires_at": None,
                         "attempts": 0,
                         "created_at": now,
                     },
-                    statement_config=statement_config,
+                    statement_config=self._queue._statement_config,
                 )
             )
             await driver.execute(SQL("SELECT pg_notify($1, $2)", channel, to_json({"event_id": event_id})))
             await driver.commit()
 
-    def _get_queue_backend(self) -> "TableEventQueue":
-        """Return the underlying TableEventQueue from the wrapper."""
-        queue_backend = self._queue._queue if has_attr(self._queue, "_queue") else None
-        if queue_backend is None:
-            msg = "Hybrid queue backend missing queue reference"
-            raise EventChannelError(msg)
-        return queue_backend
-
     async def _dequeue_with_notifies(self, connection: Any, channel: str, poll_interval: float) -> EventMessage | None:
         """Wait for a NOTIFY wakeup then dequeue from the durable table."""
         try:
@@ -181,6 +171,9 @@ async def ack(self, _event_id: str) -> None:
         """Acknowledge an event. Native notifications are fire-and-forget."""
         self._runtime.increment_metric("events.ack")
 
+    async def nack(self, _event_id: str) -> None:
+        """Return an event to the queue (no-op for native LISTEN/NOTIFY)."""
+
     async def shutdown(self) -> None:
         """Shutdown the listener connection and release resources."""
         if self._listen_connection_cm is not None:
@@ -253,7 +246,9 @@ def create_event_backend(
             except ImproperConfigurationError:
                 return None
         case "listen_notify_durable":
-            queue_backend = cast("AsyncQueueEventBackend", build_queue_backend(config, extension_settings, adapter_name="asyncpg"))
+            queue_backend = cast(
+                "AsyncTableEventQueue", build_queue_backend(config, extension_settings, adapter_name="asyncpg")
+            )
             try:
                 return AsyncpgHybridEventsBackend(config, queue_backend)
             except ImproperConfigurationError:

sqlspec/adapters/bigquery/events/store.py

@@ -60,29 +60,51 @@ def _column_types(self) -> tuple[str, str, str]:
         """
         return "JSON", "JSON", "TIMESTAMP"
 
+    def _string_type(self, length: int) -> str:
+        """Return BigQuery STRING type (length is ignored)."""
+        del length
+        return "STRING"
+
+    def _integer_type(self) -> str:
+        """Return BigQuery INT64 type."""
+        return "INT64"
+
+    def _timestamp_default(self) -> str:
+        """Return BigQuery timestamp default expression."""
+        return "CURRENT_TIMESTAMP()"
+
+    def _table_clause(self) -> str:
+        """Return BigQuery CLUSTER BY clause for query optimization."""
+        return " CLUSTER BY channel, status, available_at"
+
     def _build_create_table_sql(self) -> str:
         """Build BigQuery CREATE TABLE with CLUSTER BY optimization.
 
-        Returns:
-            DDL statement for creating the event queue table.
+        BigQuery uses CLUSTER BY for query optimization instead of indexes.
+        The clustering columns match the typical polling query pattern.
 
-        Notes:
-            BigQuery uses CLUSTER BY for query optimization instead of indexes.
-            The clustering columns match the typical polling query pattern.
+        Note: BigQuery does not support column-level PRIMARY KEY, so we
+        omit it entirely. event_id uniqueness must be enforced at insert time.
         """
+        payload_type, metadata_type, timestamp_type = self._column_types()
+        string_type = self._string_type(0)
+        integer_type = self._integer_type()
+        ts_default = self._timestamp_default()
+        table_clause = self._table_clause()
+
         return (
             f"CREATE TABLE IF NOT EXISTS {self.table_name} ("
-            "event_id STRING NOT NULL,"
-            " channel STRING NOT NULL,"
-            " payload_json JSON NOT NULL,"
-            " metadata_json JSON,"
-            " status STRING NOT NULL DEFAULT 'pending',"
-            " available_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP(),"
-            " lease_expires_at TIMESTAMP,"
-            " attempts INT64 NOT NULL DEFAULT 0,"
-            " created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP(),"
-            " acknowledged_at TIMESTAMP"
-            ") CLUSTER BY channel, status, available_at"
+            f"event_id {string_type} NOT NULL,"
+            f" channel {string_type} NOT NULL,"
+            f" payload_json {payload_type} NOT NULL,"
+            f" metadata_json {metadata_type},"
+            f" status {string_type} NOT NULL DEFAULT 'pending',"
+            f" available_at {timestamp_type} NOT NULL DEFAULT {ts_default},"
+            f" lease_expires_at {timestamp_type},"
+            f" attempts {integer_type} NOT NULL DEFAULT 0,"
+            f" created_at {timestamp_type} NOT NULL DEFAULT {ts_default},"
+            f" acknowledged_at {timestamp_type}"
+            f"){table_clause}"
         )
 
     def _build_index_sql(self) -> str | None:

sqlspec/adapters/oracledb/events/backend.py

@@ -102,6 +102,12 @@ def ack(self, _event_id: str) -> None:
         """
         self._runtime.increment_metric("events.ack")
 
+    def nack(self, _event_id: str) -> None:
+        """Return an event to the queue (no-op for Oracle AQ).
+
+        Oracle AQ does not support returning messages after commit.
+        """
+
     def shutdown(self) -> None:
         """Shutdown the backend (no-op for Oracle AQ)."""
 
@@ -183,6 +189,12 @@ async def ack(self, _event_id: str) -> None:
         """
         self._runtime.increment_metric("events.ack")
 
+    async def nack(self, _event_id: str) -> None:
+        """Return an event to the queue (no-op for Oracle AQ).
+
+        Oracle AQ does not support returning messages after commit.
+        """
+
     async def shutdown(self) -> None:
         """Shutdown the backend (no-op for Oracle AQ)."""