Database CRUD operations

The delivered pipeline touches two PostgreSQL boundaries: the backend application database and the selected client's Shopify Supabase database. The same business event can therefore create operational rows in one database and data-plane rows in another.

CRUD summary

Boundary	Create	Read	Update	Delete
Application database	Direct ShopifyEvent; SubscriberFailedEvent	Shopify connection, target database credential, due failures	Retry status, error, count, and schedule	None introduced by PR #22/#23
Client Supabase	Event and resource tables/indexes; first event and snapshot rows	No business-data reads in this pipeline	Idempotent event and snapshot upserts	No physical deletes; snapshot is_deleted is updated

Create operations

Provision target schema

ShopifyEventStorage.provision_schema() connects to the organization-selected Supabase database and creates 13 tables:

• One shopify_events event ledger.
• Twelve resource snapshot tables listed in the schema reference.
• Two indexes on the event ledger.
• One organization/topic index per resource table.

All DDL uses IF NOT EXISTS, so provisioning is repeatable. The storage path runs the same table checks for the event table and current resource table before every write.

Insert a direct webhook audit row

The HMAC-protected Django webhook endpoint performs:

ShopifyEvent.objects.create(
    organization_id=org_id,
    topic=topic,
    shop_domain=shop_domain,
    payload=payload,
)

This inserts into the application database's Shopify event model. It is not the same table as the client Supabase shopify_events table.

Insert or upsert an event ledger row

Every successfully parsed subscriber event executes an insert into the client database:

INSERT INTO shopify_events (
    event_id, organization_id, shop_domain, topic, event_type,
    resource_type, resource_id, payload, parsed_payload, received_at
)
VALUES (...)
ON CONFLICT (organization_id, shop_domain, event_id)
DO UPDATE SET
    topic = EXCLUDED.topic,
    event_type = EXCLUDED.event_type,
    resource_type = EXCLUDED.resource_type,
    resource_id = EXCLUDED.resource_id,
    payload = EXCLUDED.payload,
    parsed_payload = EXCLUDED.parsed_payload,
    received_at = EXCLUDED.received_at;

The first delivery creates the row. Redelivery with the same composite key updates its mutable event fields but preserves the original id and inserted_at.

Insert or upsert a resource snapshot

When the parser finds a resource identifier, storage inserts into the selected resource table:

INSERT INTO shopify_orders (
    organization_id, shop_domain, shopify_id, topic, event_type,
    payload, parsed_payload, is_deleted, last_seen_at
)
VALUES (...)
ON CONFLICT (organization_id, shop_domain, shopify_id)
DO UPDATE SET
    topic = EXCLUDED.topic,
    event_type = EXCLUDED.event_type,
    payload = EXCLUDED.payload,
    parsed_payload = EXCLUDED.parsed_payload,
    is_deleted = EXCLUDED.is_deleted,
    last_seen_at = EXCLUDED.last_seen_at;

The actual table identifier is composed with psycopg.sql.Identifier, not string interpolation. If no resource ID can be extracted, the event ledger row is still stored and the snapshot insert is skipped.

Insert a durable failure

When decode, parse, handler, or storage work raises, record_processing_failure() creates a SubscriberFailedEvent in the application database. It stores the enriched envelope, ownership fields where available, exception class, message, full traceback, pending status, and retry defaults.

Read operations

Resolve organization ownership

The worker reads ShopifyConnection by exact myshopify_domain. A single match produces OrgContext. No match or multiple matches causes dispatch to return without raising.

When recording a failure that lacks an explicit organization ID, the failure helper performs a lighter lookup for the first organization_id matching the shop domain.

Resolve the client database

Unless the global URL override is set, storage reads the first non-deleted Database row for the organization with:

1. type = shopify, case-insensitive; then
2. fallback database_name = shopify, case-insensitive.

The query uses select_related("credential"), then reads the one-to-one DatabaseCredential. The password is decrypted only while constructing the connection string.

Read retry candidates

retry_failed_events selects rows where:

status = pending
AND (next_retry_at IS NULL OR next_retry_at <= now)

Rows are ordered oldest first and limited to 50 by default. --event-id narrows execution to one database primary key, but the status and due-time predicates still apply.

The subscriber path does not query event or snapshot data back from the client database.

Update operations

Update through idempotent upsert

Both target tables use stable composite keys:

• Event: (organization_id, shop_domain, event_id)
• Snapshot: (organization_id, shop_domain, shopify_id)

This converts duplicate delivery and later resource events into updates instead of duplicate rows. first_seen_at remains the original snapshot insertion time; last_seen_at advances with each event.

Update failure state after retry

On a successful replay, the application database row is updated to:

status = resolved
resolved_at = current time
last_attempted_at = retry command start time

On a failed replay, it updates:

• retry_count += 1
• error_type, error_message, and traceback to the latest exception
• last_attempted_at
• next_retry_at using bounded exponential backoff
• status = failed and next_retry_at = NULL when the retry limit is reached

Delete operations

The delivered pipeline performs no SQL DELETE against event, snapshot, or failure tables.

For topics whose operation is exactly delete, the corresponding snapshot is upserted with is_deleted = true. A later create or update event for the same resource can set it back to false because the flag is part of the conflict update.

Event history and failed-event history therefore require an explicit future retention policy. Neither PR adds archival, TTL, partitioning, or purge commands.

Behavior by failure scenario

Scenario	Database result	Queue result	Recovery path
Target table missing	Storage attempts CREATE TABLE IF NOT EXISTS, then writes	ack on success	Automatic for expected schema
Expected column missing or incompatible table	Target transaction rolls back; failure row inserted in application DB	ack if failure row saved	Correct schema, then replay
Parse or code logic error	No client write; failure row inserted	ack if failure row saved	Fix code/data, then replay
Database lock/deadlock	Target transaction rolls back; exception stored	ack if failure row saved	Retry after contention clears
Target database unavailable	Connection fails; failure row inserted	ack if failure row saved	Restore database/credentials, then replay
Application failure table unavailable	Failure cannot be persisted	nack	Pub/Sub redelivery
Duplicate event delivery	Existing event and snapshot updated	ack	No operator action

Data integrity characteristics

• Event and snapshot writes are atomic within one target transaction.
• JSON payloads use PostgreSQL JSONB adapters.
• Dynamic table and index names use psycopg identifier composition.
• Composite keys provide idempotency and tenant/shop separation.
• Resource snapshot updates are last-write-wins by processing order; no Shopify sequence number comparison is implemented.
• Schema creation is idempotent but is not a versioned Supabase migration system. Structural evolution still needs explicit migration planning.