Background worker parallel impl

.github/workflows/worker_test.yaml

@@ -0,0 +1,36 @@
+name: WALRUS worker tests
+
+on:
+  pull_request:
+    branches: [master]
+  push:
+    branches: [master]
+
+jobs:
+  build:
+    name: worker tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    services:
+      postgres:
+        image: supabase/postgres:latest
+        env:
+          POSTGRES_DB: postgres
+          POSTGRES_HOST: localhost
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: password
+        ports:
+          - 5501:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: run tests
+        run: |
+          cd worker
+          cargo test --bin walrus -- --test-threads=1

.gitignore

@@ -6,3 +6,6 @@ regression.*
 __pycache__/
 *.egg-info/
 *.swp
+target/
+Cargo.lock
+.DS_Store

worker/Cargo.toml

@@ -0,0 +1,5 @@
+[workspace]
+members = [
+    "walrus",
+    "realtime",
+]

worker/README.md

@@ -0,0 +1,66 @@
+# WALRUS Worker + Realtime Transport
+
+Example showing how to stream WAL from postgres, apply row level security, and push changes to supabase realtime.
+
+- `walrus/` is responsible for receiving WAL, formatting messages, and applying row level security
+- `realtime/` is the websocket trasport layer for supabase realtime.
+
+More info about each component can be found in their directories' README.md.
+
+## Example:
+
+Requires:
+- rust/cargo
+- docker-compose
+- postgres installed locally (for `pg_recvlogical`)
+
+Clone and Navigate
+```sh
+git clone https://github.com/supabase/walrus.git
+cd walrus
+git checkout worker
+cd worker
+```
+
+Start the DB
+```sh
+docker-compose up
+```
+
+Run the `walrus` worker, piping its output to `realtime` transport
+```sh
+cargo run --bin walrus -- \
+    --connection=postgresql://postgres:password@localhost:5501/postgres --publication=walrus_pub |
+cargo run --bin realtime -- \
+    --url=wss://sendwal.fly.dev/socket \
+    --header=apikey=<apikey>
+```
+
+Connect to the database at `postgresql://postgres:password@localhost:5501/postgres`
+
+and execute the following SQL to create a subscription and a WAL record.
+
+```sql
+-- Create a table we can subscribe to
+create table book(
+    id int primary key,
+    title text
+);
+
+create publication walrus_pub for all tables;
+
+-- Create a dummy subscription to our new table
+insert into realtime.subscription(subscription_id, entity, claims)
+select
+    gen_random_uuid(),
+    'public.book',
+    jsonb_build_object(
+        'role', 'postgres',
+        'email', 'o@r.com',
+        'sub', gen_random_uuid()
+    );
+
+-- Create a record
+insert into book(id, title)
+values (1, 'Foo');
+```

worker/docker-compose.yml

@@ -0,0 +1,23 @@
+version: '3'
+services:
+  db:
+    container_name: walrus_streaming
+    build:
+      context: .
+      dockerfile: ./dockerfiles/Dockerfile
+    ports:
+      - "5501:5432"
+    command:
+      - postgres
+      - -c
+      - wal_level=logical
+      - -c
+      - fsync=off
+    healthcheck:
+      test: ["CMD-SHELL", "PGUSER=postgres", "pg_isready"]
+      interval: 1s
+      timeout: 10s
+      retries: 5
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: password

worker/dockerfiles/Dockerfile

@@ -0,0 +1,7 @@
+FROM supabase/postgres:latest
+RUN apt-get update
+RUN apt-get install build-essential postgresql-server-dev-14 -y
+
+RUN git clone https://github.com/eulerto/wal2json.git
+RUN cd wal2json && git checkout 53b548a29ebd6119323b6eb2f6013d7c5fe807ec && make && make install
+

worker/realtime/.gitignore

@@ -0,0 +1,13 @@
+.DS_Store
+.idea/
+/target
+*.iml
+**/*.rs.bk
+Cargo.lock
+*.swp
+*.rs.swp
+**/*.swp
+*.diff
+/results
+regression.diffs
+regression.out

worker/realtime/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "realtime"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+clap = { version = "3.1.12", features = ["derive"] }
+dotenv = "0.15.0"
+serde_json = "1.0"
+serde = { version = "1.0", features=["derive"] }
+uuid = { version = "1.0", features = ["serde"] }
+tokio = { version = "1", features = ["full"] }
+tokio-util = { version="0.7.2", features=["codec"] }
+tokio-tungstenite = { verison = "0.17.1", features=["native-tls"] }
+tungstenite = "0.17.1"
+futures-util = "0.3.21"
+url = "2.2.2"
+futures-channel = "0.3.21"
+futures = "0.3.21"
+log = "0.4.17"
+env_logger = "0.9.0"
+stream-reconnect = { version = "0.3", default-features = true }

worker/realtime/README.md

@@ -0,0 +1,23 @@
+# Realtime Transport
+
+Realtime Transport reads JSON form stdin and forwards it to supabase realtime
+
+See parent directory for example usage
+
+## CLI
+
+```
+realtime 0.1.0
+reads JSON from stdin and forwards it to supabase realtime
+
+USAGE:
+    realtime [OPTIONS]
+
+OPTIONS:
+    -h, --help                       Print help information
+        --header <HEADER>=<VALUE>
+        --topic <TOPIC>              [default: room:test]
+        --url <URL>                  [default: wss://sendwal.fly.dev/socket]
+    -V, --version                    Print version information
+
+```