A diet Tap option for handling atproto sync
We're flipping the P
Bailey Townsend
(Jun 30, 2026, 11:24 AM -0500)
b32e30c1
0ac9d00f
+184
-185
+1
-1
.gitignore
+1
-1
.gitignore
+60
-60
README.md
+60
-60
README.md
···
1
-
`tap`: atproto sync utility
1
+
`tab`: atproto sync utility
2
2
========================================
3
3
4
-
Tap simplifies AT sync by handling the firehose/jetstream connection, verification, backfill, and filtering. Your application connects to a Tap and receives simple JSON events for only the repos and collections you care about, no need to worry about binary formats for validating cryptographic signatures.
4
+
tab simplifies AT sync by handling the firehose/jetstream connection, verification, backfill, and filtering. Your application connects to a tab and receives simple JSON events for only the repos and collections you care about, no need to worry about binary formats for validating cryptographic signatures.
5
5
6
6
Features and design decisions:
7
7
···
14
14
- SQLite or Postgres backend
15
15
- designed for moderate scale (millions of repos, 30k+ events/sec)
16
16
17
-
There is a convenient client library in Typescript for working with Tap at [@atproto/tap](https://github.com/bluesky-social/atproto/blob/main/packages/tap/README.md)
17
+
There is a convenient client library in Typescript for working with tab at [@atproto/tab](https://github.com/bluesky-social/atproto/blob/main/packages/tab/README.md)
18
18
19
-
⚠️ Tap is still in beta and may have bugs. Please report any issues you encounter.
19
+
⚠️ tab is still in beta and may have bugs. Please report any issues you encounter.
20
20
21
21
## Quick Start
22
22
23
23
```bash
24
-
# Run tap
25
-
go run ./cmd/tap run --disable-acks=true
26
-
# By default, the service uses SQLite at `./tap.db` and binds to port `:2480`.
24
+
# Run tab
25
+
go run ./cmd/tab run --disable-acks=true
26
+
# By default, the service uses SQLite at `./tab.db` and binds to port `:2480`.
27
27
28
28
# In a separate terminal, connect to receive events:
29
29
websocat ws://localhost:2480/channel
···
38
38
39
39
## Local Development
40
40
41
-
When running Tap locally, restarts are frequent (e.g. sleep/wake, code changes). By default, Tap resumes from its saved firehose cursor on restart, which can mean replaying millions of events if the cursor is hours or days old. For local dev with a small number of tracked repos, this replay is unnecessary — any repos that changed while offline will be detected and resynced automatically.
41
+
When running tab locally, restarts are frequent (e.g. sleep/wake, code changes). By default, tab resumes from its saved firehose cursor on restart, which can mean replaying millions of events if the cursor is hours or days old. For local dev with a small number of tracked repos, this replay is unnecessary — any repos that changed while offline will be detected and resynced automatically.
42
42
43
43
Recommended local dev config:
44
44
45
45
```bash
46
-
go run ./cmd/tap run --no-replay --disable-acks
46
+
go run ./cmd/tab run --no-replay --disable-acks
47
47
```
48
48
49
49
Tips:
···
73
73
74
74
Environment variables or CLI flags:
75
75
76
-
- `TAP_DATABASE_URL`: database connection string, SQLite or PostgreSQL (default: `sqlite://./tap.db`)
77
-
- `TAP_MAX_DB_CONNS`: maximum number of database connections (default: `32`)
78
-
- `TAP_BIND`: HTTP server address (default: `:2480`)
79
-
- `TAP_PLC_URL`: PLC directory HTTP/HTTPS URL (default: `https://plc.directory`)
80
-
- `TAP_RELAY_URL`: AT Protocol relay HTTP/HTTPS URL (default: `https://relay1.us-east.bsky.network`)
81
-
- `TAP_LIGHT_RAIL_URL`: Light Rail HTTP/HTTPS URL (default: ``, but `https://lightrail.microcosm.blue` is a instance you can use)
82
-
- `TAP_JETSTREAM_URL`: If set uses the Jetstream over the firehose/relay. Other firehose settings apply to the jetstream connection (default: ``, but `https://jetstream1.us-east.bsky.network` is a public node hosted by Bluesky that you can use)
83
-
- `TAP_FIREHOSE_PARALLELISM`: concurrent firehose event processors (default: `10`)
84
-
- `TAP_RESYNC_PARALLELISM`: concurrent resync workers (default: `5`)
85
-
- `TAP_OUTBOX_PARALLELISM`: concurrent outbox workers (default: `1`)
86
-
- `TAP_CURSOR_SAVE_INTERVAL`: how often to persist upstream firehose cursor (default: `1s`)
87
-
- `TAP_NO_REPLAY`: skip saved cursor and connect to firehose head on startup; incompatible with `TAP_FULL_NETWORK`, not recommended for production (default: `false`)
88
-
- `TAP_REPO_FETCH_TIMEOUT`: timeout for fetching repo CARs from PDS (default: `300s`)
89
-
- `TAP_IDENT_CACHE_SIZE`: size of in-process identity cache (default: `2000000`)
90
-
- `TAP_OUTBOX_CAPACITY`: rough size of outbox before back pressure is applied (default: `100000`)
91
-
- `TAP_FULL_NETWORK`: track all repos on the network (default: `false`)
92
-
- `TAP_SIGNAL_COLLECTION`: track all repos with at least one record in this collection (e.g. `app.bsky.actor.profile`)
93
-
- `TAP_LIGHT_RAIL_SIGNAL_COLLECTIONS`: track all repos with at least one record in these collections. Can not be used with `TAP_SIGNAL_COLLECTION` (e.g. `community.lexicon.calendar.rsvp,community.lexicon.calendar.event`)
94
-
- `TAP_COLLECTION_FILTERS`: comma-separated collection filters, wildcards accepted (e.g., `app.bsky.feed.post,app.bsky.graph.*`)
95
-
- `TAP_DISABLE_ACKS`: fire-and-forget mode, no client acks (default: `false`)
96
-
- `TAP_WEBHOOK_URL`: webhook URL for event delivery (disables WebSocket mode)
97
-
- `TAP_OUTBOX_ONLY`: run in outbox-only mode (no firehose, resync, or enumeration) (default: `false`)
98
-
- `TAP_ADMIN_PASSWORD`: Basic auth admin password required for all requests (if set)
99
-
- `TAP_RETRY_TIMEOUT`: timeout before retrying unacked events (default: `60s`)
100
-
- `TAP_LOG_LEVEL`: log verbosity (`debug`, `info`, `warn`, `error`, default: `info`)
101
-
- `TAP_METRICS_LISTEN`: address for metrics/pprof server (disabled if empty)
76
+
- `tab_DATABASE_URL`: database connection string, SQLite or PostgreSQL (default: `sqlite://./tab.db`)
77
+
- `tab_MAX_DB_CONNS`: maximum number of database connections (default: `32`)
78
+
- `tab_BIND`: HTTP server address (default: `:2480`)
79
+
- `tab_PLC_URL`: PLC directory HTTP/HTTPS URL (default: `https://plc.directory`)
80
+
- `tab_RELAY_URL`: AT Protocol relay HTTP/HTTPS URL (default: `https://relay1.us-east.bsky.network`)
81
+
- `tab_LIGHT_RAIL_URL`: Light Rail HTTP/HTTPS URL (default: ``, but `https://lightrail.microcosm.blue` is a instance you can use)
82
+
- `tab_JETSTREAM_URL`: If set uses the Jetstream over the firehose/relay. Other firehose settings apply to the jetstream connection (default: ``, but `https://jetstream1.us-east.bsky.network` is a public node hosted by Bluesky that you can use)
83
+
- `tab_FIREHOSE_PARALLELISM`: concurrent firehose event processors (default: `10`)
84
+
- `tab_RESYNC_PARALLELISM`: concurrent resync workers (default: `5`)
85
+
- `tab_OUTBOX_PARALLELISM`: concurrent outbox workers (default: `1`)
86
+
- `tab_CURSOR_SAVE_INTERVAL`: how often to persist upstream firehose cursor (default: `1s`)
87
+
- `tab_NO_REPLAY`: skip saved cursor and connect to firehose head on startup; incompatible with `tab_FULL_NETWORK`, not recommended for production (default: `false`)
88
+
- `tab_REPO_FETCH_TIMEOUT`: timeout for fetching repo CARs from PDS (default: `300s`)
89
+
- `tab_IDENT_CACHE_SIZE`: size of in-process identity cache (default: `2000000`)
90
+
- `tab_OUTBOX_CAPACITY`: rough size of outbox before back pressure is applied (default: `100000`)
91
+
- `tab_FULL_NETWORK`: track all repos on the network (default: `false`)
92
+
- `tab_SIGNAL_COLLECTION`: track all repos with at least one record in this collection (e.g. `app.bsky.actor.profile`)
93
+
- `tab_LIGHT_RAIL_SIGNAL_COLLECTIONS`: track all repos with at least one record in these collections. Can not be used with `tab_SIGNAL_COLLECTION` (e.g. `community.lexicon.calendar.rsvp,community.lexicon.calendar.event`)
94
+
- `tab_COLLECTION_FILTERS`: comma-separated collection filters, wildcards accepted (e.g., `app.bsky.feed.post,app.bsky.graph.*`)
95
+
- `tab_DISABLE_ACKS`: fire-and-forget mode, no client acks (default: `false`)
96
+
- `tab_WEBHOOK_URL`: webhook URL for event delivery (disables WebSocket mode)
97
+
- `tab_OUTBOX_ONLY`: run in outbox-only mode (no firehose, resync, or enumeration) (default: `false`)
98
+
- `tab_ADMIN_PASSWORD`: Basic auth admin password required for all requests (if set)
99
+
- `tab_RETRY_TIMEOUT`: timeout before retrying unacked events (default: `60s`)
100
+
- `tab_LOG_LEVEL`: log verbosity (`debug`, `info`, `warn`, `error`, default: `info`)
101
+
- `tab_METRICS_LISTEN`: address for metrics/pprof server (disabled if empty)
102
102
103
103
## Delivery Modes
104
104
105
-
Tap supports three delivery modes:
105
+
tab supports three delivery modes:
106
106
107
-
**WebSocket with acks** (default): Client sends acks each event once it has been processed/persisted. Ensures that no data is lost and client does not need to handle cursors. It's recommended to use a client library such as [@atproto/tap](https://github.com/bluesky-social/atproto/tree/main/packages/tap/README.md) when using this mode.
107
+
**WebSocket with acks** (default): Client sends acks each event once it has been processed/persisted. Ensures that no data is lost and client does not need to handle cursors. It's recommended to use a client library such as [@atproto/tab](https://github.com/bluesky-social/atproto/tree/main/packages/tab/README.md) when using this mode.
108
108
109
-
**Fire-and-forget**: Set `TAP_DISABLE_ACKS=true`. Events are sent and considered "acked" once the client receives them. Simpler but may result in data loss. Recommended for testing purposes or when data integrity is not critical.
109
+
**Fire-and-forget**: Set `tab_DISABLE_ACKS=true`. Events are sent and considered "acked" once the client receives them. Simpler but may result in data loss. Recommended for testing purposes or when data integrity is not critical.
110
110
111
-
**Webhook**: Set `TAP_WEBHOOK_URL=http://...`. Events are POSTed as JSON. Events considered "acked" once the webhook responds with a 200. Recommended for lower throughput serverless environments.
111
+
**Webhook**: Set `tab_WEBHOOK_URL=http://...`. Events are POSTed as JSON. Events considered "acked" once the webhook responds with a 200. Recommended for lower throughput serverless environments.
112
112
113
113
114
114
## Network Boundary Modes
115
115
116
-
Tap syncs a subset of repos in the network. It can operate in three modes for determining this network boundary.
116
+
tab syncs a subset of repos in the network. It can operate in three modes for determining this network boundary.
117
117
118
-
**Dynamically Configured** (default): Tap starts out tracking no repos. Specific repos can then by added via `/repos/add` and removed via `/repos/remove`.
118
+
**Dynamically Configured** (default): tab starts out tracking no repos. Specific repos can then by added via `/repos/add` and removed via `/repos/remove`.
119
119
120
-
**Collection Signal**: Set `TAP_SIGNAL_COLLECTION=com.example.nsid`. Track all repos that have at least one record in the specified collection. Many applications create a "declaration" or "profile" in a repo when that repo uses that application
120
+
**Collection Signal**: Set `tab_SIGNAL_COLLECTION=com.example.nsid`. Track all repos that have at least one record in the specified collection. Many applications create a "declaration" or "profile" in a repo when that repo uses that application
121
121
122
-
**Full Network**: Set `TAP_FULL_NETWORK=true`. Enumerates and tracks all findable repos on the entire network. Resource-intensive and takes days/weeks to complete backfill.
122
+
**Full Network**: Set `tab_FULL_NETWORK=true`. Enumerates and tracks all findable repos on the entire network. Resource-intensive and takes days/weeks to complete backfill.
123
123
124
124
## Collection Filtering
125
125
126
-
After narrowing down the network to a subset of repos, Tap can further filter records down to a specified set of collections. Filters apply to record events only. Identity events are always delivered for tracked repos.
126
+
After narrowing down the network to a subset of repos, tab can further filter records down to a specified set of collections. Filters apply to record events only. Identity events are always delivered for tracked repos.
127
127
128
-
If you are interested syncing all of a single record type, it is important to specify that collection as both the signal collection and the filter collection. For example: `TAP_SIGNAL_COLLECTION=com.example.nsid TAP_COLLECTION_FILTERS=com.example.nsid`
128
+
If you are interested syncing all of a single record type, it is important to specify that collection as both the signal collection and the filter collection. For example: `tab_SIGNAL_COLLECTION=com.example.nsid tab_COLLECTION_FILTERS=com.example.nsid`
129
129
130
130
Collection filters use wildcards but only at the period breaks in NSIDs. For example:
131
131
132
-
`TAP_COLLECTION_FILTERS=app.bsky.feed.post,app.bsky.graph.*`
132
+
`tab_COLLECTION_FILTERS=app.bsky.feed.post,app.bsky.graph.*`
133
133
134
134
135
135
## Event Format
···
178
178
179
179
When a repo is added (via `/repos/add`, full network mode, or collection discovery):
180
180
181
-
1. **Historical backfill**: Tap fetches the full repo from the account's PDS using `com.atproto.sync.getRepo`
181
+
1. **Historical backfill**: tab fetches the full repo from the account's PDS using `com.atproto.sync.getRepo`
182
182
2. **Live event buffering**: Any firehose events for this repo during backfill are held in memory
183
183
3. **Ordering guarantee**: Historical events (marked `live: false`) are delivered first
184
184
4. **Cutover**: After historical events complete, buffered live events are drained
···
188
188
189
189
### Per-Repo Ordering Rules
190
190
191
-
Tap offloads cursor management and takes care of delivery guarantees. Events are delivered *at least once*. Events may be delivered more than once if Tap crashes and restarts before receiving an ack for a given event or if the event times out before being acked (default 10s).
191
+
tab offloads cursor management and takes care of delivery guarantees. Events are delivered *at least once*. Events may be delivered more than once if tab crashes and restarts before receiving an ack for a given event or if the event times out before being acked (default 10s).
192
192
193
-
There is no global ordering of events across repos. However Tap will ensure ordering within each repo and will avoid sending the next event until the previous event has completed processing.
193
+
There is no global ordering of events across repos. However tab will ensure ordering within each repo and will avoid sending the next event until the previous event has completed processing.
194
194
195
195
Events for the same repo are delivered with strict ordering:
196
196
···
208
208
209
209
## Authentication
210
210
211
-
If exposing Tap to the internet, you should set `TAP_ADMIN_PASSWORD` to require authentication for all API requests.
211
+
If exposing tab to the internet, you should set `tab_ADMIN_PASSWORD` to require authentication for all API requests.
212
212
213
-
Tap uses HTTP Basic authentication with the username `admin`. Basic auth works by concatenating the username and password with a colon (`admin:yourpassword`), then base64-encoding the result. This is sent in the `Authorization` header as `Basic <encoded-value>`.
213
+
tab uses HTTP Basic authentication with the username `admin`. Basic auth works by concatenating the username and password with a colon (`admin:yourpassword`), then base64-encoding the result. This is sent in the `Authorization` header as `Basic <encoded-value>`.
214
214
215
215
Example with curl:
216
216
```bash
···
219
219
-d '{"dids": ["did:plc:..."]}'
220
220
```
221
221
222
-
When using webhook mode, Tap sends the same Basic auth credentials to your webhook endpoint.
222
+
When using webhook mode, tab sends the same Basic auth credentials to your webhook endpoint.
223
223
224
224
## Operations
225
225
226
-
Tap logs to stdout in JSON format. The firehose consumer automatically reconnects with exponential backoff on relay failures. Cursor position is saved periodically (default 1 second) and restored on restart.
226
+
tab logs to stdout in JSON format. The firehose consumer automatically reconnects with exponential backoff on relay failures. Cursor position is saved periodically (default 1 second) and restored on restart.
227
227
228
228
SQLite is tuned for high write throughput: WAL mode, 10-second busy timeout, `synchronous=NORMAL`, 64MB cache, batched deletes. The outbox buffers up to 1M pending events in memory.
229
229
···
233
233
234
234
## Distribution & Deployment
235
235
236
-
Tap is distributed as a single Go binary and is easy to build and run.
236
+
tab is distributed as a single Go binary and is easy to build and run.
237
237
238
238
**Build from source:**
239
239
```bash
240
-
go build -o tap ./cmd/tap
241
-
./tap run
240
+
go build -o tab ./cmd/tab
241
+
./tab run
242
242
```
243
243
244
244
**Docker:**
245
245
246
246
A pre-built Docker image is also available:
247
247
```bash
248
-
docker pull ghcr.io/bluesky-social/indigo/tap:latest
249
-
docker run -p 2480:2480 ghcr.io/bluesky-social/indigo/tap:latest
248
+
docker pull ghcr.io/bluesky-social/indigo/tab:latest
249
+
docker run -p 2480:2480 ghcr.io/bluesky-social/indigo/tab:latest
250
250
```
251
251
252
252
To persist data, mount a volume at `/data`:
253
253
```bash
254
-
docker run -p 2480:2480 -v ./data:/data ghcr.io/bluesky-social/indigo/tap:latest
254
+
docker run -p 2480:2480 -v ./data:/data ghcr.io/bluesky-social/indigo/tab:latest
255
255
```
256
256
257
-
The Dockerfile is included in the repo at `cmd/tap/Dockerfile` if you need to build your own image.
257
+
The Dockerfile is included in the repo at `cmd/tab/Dockerfile` if you need to build your own image.
+5
-5
cmd/tap/Dockerfile
cmd/tab/Dockerfile
+5
-5
cmd/tap/Dockerfile
cmd/tab/Dockerfile
···
1
1
# Run this dockerfile from the top level of the indigo git repository like:
2
2
#
3
-
# podman build -f ./cmd/tap/Dockerfile -t tap .
3
+
# podman build -f ./cmd/tab/Dockerfile -t tab .
4
4
5
5
### Compile stage
6
6
FROM golang:1.26-alpine3.22 AS build-env
···
11
11
12
12
# timezone data for alpine builds
13
13
RUN GIT_VERSION=$(git describe --tags --long --always) && \
14
-
go build -tags timetzdata -o /tap ./cmd/tap
14
+
go build -tags timetzdata -o /tab ./cmd/tab
15
15
16
16
### Run stage
17
17
FROM alpine:3.22
···
21
21
22
22
RUN mkdir -p /data
23
23
WORKDIR /data
24
-
COPY --from=build-env /tap /
24
+
COPY --from=build-env /tab /
25
25
26
26
# small things to make golang binaries work well under alpine
27
27
ENV GODEBUG=netdns=go
···
29
29
30
30
EXPOSE 2480
31
31
32
-
CMD ["/tap", "run"]
32
+
CMD ["/tab", "run"]
33
33
34
34
LABEL org.opencontainers.image.source=https://github.com/bluesky-social/indigo
35
-
LABEL org.opencontainers.image.description="AT Protocol Sync Utility (tap)"
35
+
LABEL org.opencontainers.image.description="AT Protocol Sync Utility (tab)"
36
36
LABEL org.opencontainers.image.licenses=MIT
+3
-4
cmd/tap/crawler.go
cmd/tab/crawler.go
+3
-4
cmd/tap/crawler.go
cmd/tab/crawler.go
···
8
8
"strings"
9
9
"time"
10
10
11
-
lexutil "github.com/bluesky-social/indigo/lex/util"
12
-
13
11
comatproto "github.com/bluesky-social/indigo/api/atproto"
14
12
"github.com/bluesky-social/indigo/atproto/atclient"
15
-
"github.com/bluesky-social/indigo/cmd/tap/models"
13
+
lexutil "github.com/bluesky-social/indigo/lex/util"
16
14
"go.opentelemetry.io/otel/attribute"
17
15
"gorm.io/gorm"
18
16
"gorm.io/gorm/clause"
17
+
"tangled.org/pds.dad/tab/cmd/tab/models"
19
18
)
20
19
21
20
type Crawler struct {
···
28
27
lightRailSignalCollections []string
29
28
}
30
29
31
-
func NewCrawler(logger *slog.Logger, db *gorm.DB, config *TapConfig) *Crawler {
30
+
func NewCrawler(logger *slog.Logger, db *gorm.DB, config *tabConfig) *Crawler {
32
31
url := config.RelayUrl
33
32
if config.LightRailUrl != "" {
34
33
url = config.LightRailUrl
+1
-1
cmd/tap/db_helpers.go
cmd/tab/db_helpers.go
+1
-1
cmd/tap/db_helpers.go
cmd/tab/db_helpers.go
+2
-2
cmd/tap/event_manager.go
cmd/tab/event_manager.go
+2
-2
cmd/tap/event_manager.go
cmd/tab/event_manager.go
···
8
8
"sync/atomic"
9
9
"time"
10
10
11
-
"github.com/bluesky-social/indigo/cmd/tap/models"
12
11
"gorm.io/gorm"
13
12
"gorm.io/gorm/clause"
13
+
"tangled.org/pds.dad/tab/cmd/tab/models"
14
14
)
15
15
16
16
type EventManager struct {
···
28
28
pendingIDs chan uint
29
29
}
30
30
31
-
func NewEventManager(logger *slog.Logger, db *gorm.DB, config *TapConfig) *EventManager {
31
+
func NewEventManager(logger *slog.Logger, db *gorm.DB, config *tabConfig) *EventManager {
32
32
return &EventManager{
33
33
logger: logger.With("component", "event_manager"),
34
34
db: db,
+3
-3
cmd/tap/firehose.go
cmd/tab/firehose.go
+3
-3
cmd/tap/firehose.go
cmd/tab/firehose.go
···
13
13
"github.com/bluesky-social/indigo/atproto/atdata"
14
14
"github.com/bluesky-social/indigo/atproto/repo"
15
15
"github.com/bluesky-social/indigo/atproto/syntax"
16
-
"github.com/bluesky-social/indigo/cmd/tap/models"
17
16
"github.com/bluesky-social/indigo/events"
18
17
"github.com/bluesky-social/indigo/events/schedulers/parallel"
19
18
"github.com/gorilla/websocket"
20
19
"go.opentelemetry.io/otel"
21
20
"gorm.io/gorm"
21
+
"tangled.org/pds.dad/tab/cmd/tab/models"
22
22
)
23
23
24
-
var tracer = otel.Tracer("tap")
24
+
var tracer = otel.Tracer("tab")
25
25
26
26
type FirehoseProcessor struct {
27
27
logger *slog.Logger
···
42
42
lastSeq atomic.Int64
43
43
}
44
44
45
-
func NewFirehoseProcessor(logger *slog.Logger, db *gorm.DB, events *EventManager, repos *RepoManager, config *TapConfig) *FirehoseProcessor {
45
+
func NewFirehoseProcessor(logger *slog.Logger, db *gorm.DB, events *EventManager, repos *RepoManager, config *tabConfig) *FirehoseProcessor {
46
46
return &FirehoseProcessor{
47
47
logger: logger.With("component", "firehose"),
48
48
db: db,
+3
-3
cmd/tap/firehose_test.go
cmd/tab/firehose_test.go
+3
-3
cmd/tap/firehose_test.go
cmd/tab/firehose_test.go
···
7
7
comatproto "github.com/bluesky-social/indigo/api/atproto"
8
8
"github.com/bluesky-social/indigo/atproto/identity"
9
9
"github.com/bluesky-social/indigo/atproto/syntax"
10
-
"github.com/bluesky-social/indigo/cmd/tap/models"
10
+
"tangled.org/pds.dad/tab/cmd/tab/models"
11
11
)
12
12
13
13
func newTestFirehoseProcessor(te *testEnv, fullNetwork bool) *FirehoseProcessor {
14
-
config := &TapConfig{
14
+
config := &tabConfig{
15
15
RelayUrl: "wss://relay.test.example",
16
16
FullNetworkMode: fullNetwork,
17
17
FirehoseParallelism: 1,
···
550
550
551
551
makeProcessor := func(te *testEnv, disableReplay bool) *FirehoseProcessor {
552
552
t.Helper()
553
-
config := &TapConfig{
553
+
config := &tabConfig{
554
554
RelayUrl: relayUrl,
555
555
FirehoseParallelism: 1,
556
556
FirehoseCursorSaveInterval: 5 * time.Second,
+4
-4
cmd/tap/jetstream.go
cmd/tab/jetstream.go
+4
-4
cmd/tap/jetstream.go
cmd/tab/jetstream.go
···
12
12
13
13
"slices"
14
14
15
-
"github.com/bluesky-social/indigo/cmd/tap/models"
16
15
"github.com/gorilla/websocket"
17
16
"go.opentelemetry.io/otel"
18
17
"gorm.io/gorm"
19
-
"tangled.org/pds.dad/tab/cmd/tap/jetstream"
18
+
"tangled.org/pds.dad/tab/cmd/tab/jetstream"
19
+
"tangled.org/pds.dad/tab/cmd/tab/models"
20
20
)
21
21
22
-
var jetstreamTracer = otel.Tracer("tap")
22
+
var jetstreamTracer = otel.Tracer("tab")
23
23
24
24
type JetstreamProcessor struct {
25
25
logger *slog.Logger
···
39
39
lastTimeUs atomic.Int64
40
40
}
41
41
42
-
func NewJetstreamProcessor(logger *slog.Logger, db *gorm.DB, events *EventManager, repos *RepoManager, config *TapConfig) *JetstreamProcessor {
42
+
func NewJetstreamProcessor(logger *slog.Logger, db *gorm.DB, events *EventManager, repos *RepoManager, config *tabConfig) *JetstreamProcessor {
43
43
return &JetstreamProcessor{
44
44
logger: logger.With("component", "jetstream"),
45
45
db: db,
cmd/tap/jetstream/consumer.go
cmd/tab/jetstream/consumer.go
cmd/tap/jetstream/consumer.go
cmd/tab/jetstream/consumer.go
cmd/tap/jetstream/metrics.go
cmd/tab/jetstream/metrics.go
cmd/tap/jetstream/metrics.go
cmd/tab/jetstream/metrics.go
cmd/tap/jetstream/scheduler.go
cmd/tab/jetstream/scheduler.go
cmd/tap/jetstream/scheduler.go
cmd/tab/jetstream/scheduler.go
cmd/tap/jetstream/types.go
cmd/tab/jetstream/types.go
cmd/tap/jetstream/types.go
cmd/tab/jetstream/types.go
cmd/tap/jetstream/zstd_dictionary
cmd/tab/jetstream/zstd_dictionary
cmd/tap/jetstream/zstd_dictionary
cmd/tab/jetstream/zstd_dictionary
+41
-41
cmd/tap/main.go
cmd/tab/main.go
+41
-41
cmd/tap/main.go
cmd/tab/main.go
···
32
32
defer cancel()
33
33
34
34
app := &cli.Command{
35
-
Name: "tap",
35
+
Name: "tab",
36
36
Usage: "atproto sync tool",
37
37
Version: versioninfo.Short(),
38
38
Commands: []*cli.Command{
39
39
{
40
40
Name: "run",
41
-
Usage: "start the tap server",
41
+
Usage: "start the tab server",
42
42
Flags: []cli.Flag{
43
43
&cli.StringFlag{
44
44
Name: "env",
45
45
Usage: "environment name for observability",
46
46
Value: "dev",
47
-
Sources: cli.EnvVars("TAP_ENV"),
47
+
Sources: cli.EnvVars("tab_ENV"),
48
48
},
49
49
&cli.StringFlag{
50
50
Name: "db-url",
51
51
Usage: "database connection string (sqlite://path or postgres://...)",
52
-
Value: "sqlite://./tap.db",
53
-
Sources: cli.EnvVars("TAP_DATABASE_URL"),
52
+
Value: "sqlite://./tab.db",
53
+
Sources: cli.EnvVars("tab_DATABASE_URL"),
54
54
},
55
55
&cli.IntFlag{
56
56
Name: "max-db-conn",
57
57
Usage: "maximum number of database connections",
58
58
Value: 32,
59
-
Sources: cli.EnvVars("TAP_MAX_DB_CONNS"),
59
+
Sources: cli.EnvVars("tab_MAX_DB_CONNS"),
60
60
},
61
61
&cli.StringFlag{
62
62
Name: "bind",
63
63
Usage: "address and port to listen on for HTTP APIs",
64
64
Value: ":2480",
65
-
Sources: cli.EnvVars("TAP_BIND"),
65
+
Sources: cli.EnvVars("tab_BIND"),
66
66
},
67
67
&cli.StringFlag{
68
68
Name: "plc-url",
69
69
Usage: "PLC registry HTTP/HTTPS url",
70
70
Value: "https://plc.directory",
71
-
Sources: cli.EnvVars("TAP_PLC_URL", "ATP_PLC_HOST"),
71
+
Sources: cli.EnvVars("tab_PLC_URL", "ATP_PLC_HOST"),
72
72
},
73
73
&cli.StringFlag{
74
74
Name: "relay-url",
75
75
Usage: "AT Protocol relay HTTP/HTTPS url",
76
76
Value: "https://relay1.us-east.bsky.network",
77
-
Sources: cli.EnvVars("TAP_RELAY_URL"),
77
+
Sources: cli.EnvVars("tab_RELAY_URL"),
78
78
},
79
79
&cli.StringFlag{
80
80
Name: "light-rail-url",
81
81
Usage: "Microcosm Lightrail HTTP/HTTPS url",
82
82
Value: "",
83
-
Sources: cli.EnvVars("TAP_LIGHT_RAIL_URL"),
83
+
Sources: cli.EnvVars("tab_LIGHT_RAIL_URL"),
84
84
},
85
85
&cli.StringFlag{
86
86
Name: "jetstream-url",
87
87
Usage: "JetStream HTTP/HTTPS url",
88
-
Sources: cli.EnvVars("TAP_JETSTREAM_URL"),
88
+
Sources: cli.EnvVars("tab_JETSTREAM_URL"),
89
89
},
90
90
&cli.IntFlag{
91
91
Name: "firehose-parallelism",
92
92
Usage: "number of parallel firehose event processors",
93
93
Value: 10,
94
-
Sources: cli.EnvVars("TAP_FIREHOSE_PARALLELISM"),
94
+
Sources: cli.EnvVars("tab_FIREHOSE_PARALLELISM"),
95
95
},
96
96
&cli.IntFlag{
97
97
Name: "resync-parallelism",
98
98
Usage: "number of parallel resync workers",
99
99
Value: 5,
100
-
Sources: cli.EnvVars("TAP_RESYNC_PARALLELISM"),
100
+
Sources: cli.EnvVars("tab_RESYNC_PARALLELISM"),
101
101
},
102
102
&cli.IntFlag{
103
103
Name: "outbox-parallelism",
104
104
Usage: "number of parallel outbox workers",
105
105
Value: 1,
106
-
Sources: cli.EnvVars("TAP_OUTBOX_PARALLELISM"),
106
+
Sources: cli.EnvVars("tab_OUTBOX_PARALLELISM"),
107
107
},
108
108
&cli.DurationFlag{
109
109
Name: "cursor-save-interval",
110
110
Usage: "how often to save firehose cursor",
111
111
Value: 1 * time.Second,
112
-
Sources: cli.EnvVars("TAP_CURSOR_SAVE_INTERVAL"),
112
+
Sources: cli.EnvVars("tab_CURSOR_SAVE_INTERVAL"),
113
113
},
114
114
&cli.BoolFlag{
115
115
Name: "no-replay",
116
116
Usage: "skip saved cursor and connect to firehose head on startup (incompatible with --full-network, not recommended for production)",
117
-
Sources: cli.EnvVars("TAP_NO_REPLAY"),
117
+
Sources: cli.EnvVars("tab_NO_REPLAY"),
118
118
},
119
119
&cli.DurationFlag{
120
120
Name: "repo-fetch-timeout",
121
121
Usage: "timeout when fetching repo CARs from PDS (e.g. 180s)",
122
122
Value: 300 * time.Second,
123
-
Sources: cli.EnvVars("TAP_REPO_FETCH_TIMEOUT"),
123
+
Sources: cli.EnvVars("tab_REPO_FETCH_TIMEOUT"),
124
124
},
125
125
&cli.IntFlag{
126
126
Name: "ident-cache-size",
···
132
132
Name: "outbox-capacity",
133
133
Usage: "rough size of outbox before back pressure is applied",
134
134
Value: 100_000,
135
-
Sources: cli.EnvVars("TAP_OUTBOX_CAPACITY"),
135
+
Sources: cli.EnvVars("tab_OUTBOX_CAPACITY"),
136
136
},
137
137
&cli.BoolFlag{
138
138
Name: "full-network",
139
139
Usage: "enumerate and sync all repos on the network",
140
-
Sources: cli.EnvVars("TAP_FULL_NETWORK"),
140
+
Sources: cli.EnvVars("tab_FULL_NETWORK"),
141
141
},
142
142
&cli.StringFlag{
143
143
Name: "signal-collection",
144
144
Usage: "enumerate repos by collection (exact NSID)",
145
-
Sources: cli.EnvVars("TAP_SIGNAL_COLLECTION"),
145
+
Sources: cli.EnvVars("tab_SIGNAL_COLLECTION"),
146
146
},
147
147
&cli.StringSliceFlag{
148
148
Name: "light-rail-signal-collections",
149
149
Usage: "enumerate repos by collections (exact NSIDs)",
150
-
Sources: cli.EnvVars("TAP_LIGHT_RAIL_SIGNAL_COLLECTIONS"),
150
+
Sources: cli.EnvVars("tab_LIGHT_RAIL_SIGNAL_COLLECTIONS"),
151
151
},
152
152
&cli.BoolFlag{
153
153
Name: "disable-acks",
154
154
Usage: "disable client acknowledgments (fire-and-forget mode)",
155
-
Sources: cli.EnvVars("TAP_DISABLE_ACKS"),
155
+
Sources: cli.EnvVars("tab_DISABLE_ACKS"),
156
156
},
157
157
&cli.StringFlag{
158
158
Name: "webhook-url",
159
159
Usage: "webhook URL for event delivery (instead of WebSocket)",
160
-
Sources: cli.EnvVars("TAP_WEBHOOK_URL"),
160
+
Sources: cli.EnvVars("tab_WEBHOOK_URL"),
161
161
},
162
162
&cli.StringSliceFlag{
163
163
Name: "collection-filters",
164
164
Usage: "filter output records by collection (supports wildcards)",
165
-
Sources: cli.EnvVars("TAP_COLLECTION_FILTERS"),
165
+
Sources: cli.EnvVars("tab_COLLECTION_FILTERS"),
166
166
},
167
167
&cli.BoolFlag{
168
168
Name: "outbox-only",
169
169
Usage: "run in outbox-only mode (no firehose, resync, or enumeration)",
170
-
Sources: cli.EnvVars("TAP_OUTBOX_ONLY"),
170
+
Sources: cli.EnvVars("tab_OUTBOX_ONLY"),
171
171
},
172
172
&cli.StringFlag{
173
173
Name: "admin-password",
174
174
Usage: "Basic auth admin password required for all requests (if set)",
175
-
Sources: cli.EnvVars("TAP_ADMIN_PASSWORD"),
175
+
Sources: cli.EnvVars("tab_ADMIN_PASSWORD"),
176
176
},
177
177
&cli.DurationFlag{
178
178
Name: "retry-timeout",
179
179
Usage: "timeout before retrying unacked events",
180
180
Value: 60 * time.Second,
181
-
Sources: cli.EnvVars("TAP_RETRY_TIMEOUT"),
181
+
Sources: cli.EnvVars("tab_RETRY_TIMEOUT"),
182
182
},
183
183
&cli.StringFlag{
184
184
Name: "log-level",
185
185
Usage: "log verbosity level (debug, info, warn, error)",
186
186
Value: "info",
187
-
Sources: cli.EnvVars("TAP_LOG_LEVEL", "LOG_LEVEL"),
187
+
Sources: cli.EnvVars("tab_LOG_LEVEL", "LOG_LEVEL"),
188
188
},
189
189
&cli.StringFlag{
190
190
Name: "metrics-listen",
191
191
Usage: "address for metrics/pprof server (disabled if empty)",
192
-
Sources: cli.EnvVars("TAP_METRICS_LISTEN"),
192
+
Sources: cli.EnvVars("tab_METRICS_LISTEN"),
193
193
},
194
194
},
195
-
Action: runTap,
195
+
Action: runtab,
196
196
},
197
197
},
198
198
}
···
200
200
return app.Run(ctx, args)
201
201
}
202
202
203
-
func runTap(ctx context.Context, cmd *cli.Command) error {
203
+
func runtab(ctx context.Context, cmd *cli.Command) error {
204
204
logger := configLogger(cmd, os.Stdout)
205
205
slog.SetDefault(logger)
206
206
···
245
245
return fmt.Errorf("--light-rail-url cannot be used with --signal-collection or --full-network")
246
246
}
247
247
248
-
config := TapConfig{
248
+
config := tabConfig{
249
249
DatabaseURL: cmd.String("db-url"),
250
250
DBMaxConns: int(cmd.Int("max-db-conn")),
251
251
PLCURL: plcUrl,
···
271
271
RetryTimeout: cmd.Duration("retry-timeout"),
272
272
}
273
273
274
-
logger.Info("creating tap service")
275
-
tap, err := NewTap(config)
274
+
logger.Info("creating tab service")
275
+
tab, err := Newtab(config)
276
276
if err != nil {
277
277
return err
278
278
}
279
279
280
280
if !config.OutboxOnly {
281
-
go tap.crawler.Run(ctx)
281
+
go tab.crawler.Run(ctx)
282
282
}
283
283
284
284
svcErr := make(chan error, 1)
···
287
287
go func() {
288
288
if config.JetstreamUrl != "" {
289
289
logger.Info("starting jetstream consumer", "url", config.JetstreamUrl)
290
-
if err := tap.jetstream.Run(ctx); err != nil {
290
+
if err := tab.jetstream.Run(ctx); err != nil {
291
291
svcErr <- err
292
292
}
293
293
} else {
294
294
logger.Info("starting firehose consumer")
295
-
if err := tap.firehose.Run(ctx); err != nil {
295
+
if err := tab.firehose.Run(ctx); err != nil {
296
296
svcErr <- err
297
297
}
298
298
}
299
299
}()
300
300
}
301
301
302
-
go tap.Run(ctx)
302
+
go tab.Run(ctx)
303
303
304
304
go func() {
305
305
logger.Info("starting HTTP server", "addr", cmd.String("bind"))
306
-
if err := tap.server.Start(cmd.String("bind")); err != nil {
306
+
if err := tab.server.Start(cmd.String("bind")); err != nil {
307
307
svcErr <- err
308
308
}
309
309
}()
···
334
334
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
335
335
defer shutdownCancel()
336
336
337
-
if err := tap.server.Shutdown(shutdownCtx); err != nil {
337
+
if err := tab.server.Shutdown(shutdownCtx); err != nil {
338
338
logger.Error("error during shutdown", "error", err)
339
339
return err
340
340
}
341
341
342
-
if err := tap.CloseDb(shutdownCtx); err != nil {
342
+
if err := tab.CloseDb(shutdownCtx); err != nil {
343
343
return err
344
344
}
345
345
+17
-17
cmd/tap/metrics.go
cmd/tab/metrics.go
+17
-17
cmd/tap/metrics.go
cmd/tab/metrics.go
···
8
8
// Firehose metrics
9
9
var (
10
10
firehoseEventsReceived = promauto.NewCounter(prometheus.CounterOpts{
11
-
Name: "tap_firehose_events_received_total",
11
+
Name: "tab_firehose_events_received_total",
12
12
Help: "Total number of events received from firehose",
13
13
})
14
14
firehoseEventsProcessed = promauto.NewCounter(prometheus.CounterOpts{
15
-
Name: "tap_firehose_events_processed_total",
15
+
Name: "tab_firehose_events_processed_total",
16
16
Help: "Total number of events successfully processed",
17
17
})
18
18
firehoseEventsSkipped = promauto.NewCounter(prometheus.CounterOpts{
19
-
Name: "tap_firehose_events_skipped_total",
19
+
Name: "tab_firehose_events_skipped_total",
20
20
Help: "Total number of events skipped (not tracked)",
21
21
})
22
22
firehoseLastSeq = promauto.NewGauge(prometheus.GaugeOpts{
23
-
Name: "tap_firehose_last_seq",
23
+
Name: "tab_firehose_last_seq",
24
24
Help: "Sequence number of last processed firehose event",
25
25
})
26
26
)
···
28
28
// Jetstream metrics
29
29
var (
30
30
jetstreamEventsReceived = promauto.NewCounter(prometheus.CounterOpts{
31
-
Name: "tap_jetstream_events_received_total",
31
+
Name: "tab_jetstream_events_received_total",
32
32
Help: "Total number of events received from jetstream",
33
33
})
34
34
jetstreamEventsProcessed = promauto.NewCounter(prometheus.CounterOpts{
35
-
Name: "tap_jetstream_events_processed_total",
35
+
Name: "tab_jetstream_events_processed_total",
36
36
Help: "Total number of events successfully processed",
37
37
})
38
38
jetstreamEventsSkipped = promauto.NewCounter(prometheus.CounterOpts{
39
-
Name: "tap_jetstream_events_skipped_total",
39
+
Name: "tab_jetstream_events_skipped_total",
40
40
Help: "Total number of events skipped (not tracked)",
41
41
})
42
42
jetstreamLastSeq = promauto.NewGauge(prometheus.GaugeOpts{
43
-
Name: "tap_jetstream_last_seq",
43
+
Name: "tab_jetstream_last_seq",
44
44
Help: "Sequence number of last processed jetstream event",
45
45
})
46
46
)
···
48
48
// Buffer metrics
49
49
var (
50
50
eventCacheSize = promauto.NewGauge(prometheus.GaugeOpts{
51
-
Name: "tap_event_cache_size",
51
+
Name: "tab_event_cache_size",
52
52
Help: "Number of events in memory cache",
53
53
})
54
54
)
···
56
56
// Resync metrics
57
57
var (
58
58
resyncsStarted = promauto.NewCounter(prometheus.CounterOpts{
59
-
Name: "tap_resyncs_started_total",
59
+
Name: "tab_resyncs_started_total",
60
60
Help: "Total number of repo resyncs started",
61
61
})
62
62
resyncsCompleted = promauto.NewCounter(prometheus.CounterOpts{
63
-
Name: "tap_resyncs_completed_total",
63
+
Name: "tab_resyncs_completed_total",
64
64
Help: "Total number of repo resyncs completed",
65
65
})
66
66
resyncsFailed = promauto.NewCounter(prometheus.CounterOpts{
67
-
Name: "tap_resyncs_failed_total",
67
+
Name: "tab_resyncs_failed_total",
68
68
Help: "Total number of repo resyncs failed",
69
69
})
70
70
resyncDuration = promauto.NewHistogram(prometheus.HistogramOpts{
71
-
Name: "tap_resync_duration_seconds",
71
+
Name: "tab_resync_duration_seconds",
72
72
Help: "Duration of repo resync operations",
73
73
Buckets: prometheus.ExponentialBuckets(0.1, 2, 12),
74
74
})
···
77
77
// Outbox delivery metrics
78
78
var (
79
79
eventsDelivered = promauto.NewCounter(prometheus.CounterOpts{
80
-
Name: "tap_events_delivered_total",
80
+
Name: "tab_events_delivered_total",
81
81
Help: "Total number of events delivered to clients",
82
82
})
83
83
eventsAcked = promauto.NewCounter(prometheus.CounterOpts{
84
-
Name: "tap_events_acked_total",
84
+
Name: "tab_events_acked_total",
85
85
Help: "Total number of events acknowledged",
86
86
})
87
87
webhookRequests = promauto.NewCounterVec(prometheus.CounterOpts{
88
-
Name: "tap_webhook_requests_total",
88
+
Name: "tab_webhook_requests_total",
89
89
Help: "Total webhook requests by status",
90
90
}, []string{"status"})
91
91
)
···
93
93
// Crawler metrics
94
94
var (
95
95
crawlerReposDiscovered = promauto.NewCounter(prometheus.CounterOpts{
96
-
Name: "tap_crawler_repos_discovered_total",
96
+
Name: "tab_crawler_repos_discovered_total",
97
97
Help: "Total repos discovered by crawler",
98
98
})
99
99
)
cmd/tap/models/models.go
cmd/tab/models/models.go
cmd/tap/models/models.go
cmd/tab/models/models.go
+1
-1
cmd/tap/outbox.go
cmd/tab/outbox.go
+1
-1
cmd/tap/outbox.go
cmd/tab/outbox.go
···
44
44
ctx context.Context
45
45
}
46
46
47
-
func NewOutbox(logger *slog.Logger, events *EventManager, config *TapConfig) *Outbox {
47
+
func NewOutbox(logger *slog.Logger, events *EventManager, config *tabConfig) *Outbox {
48
48
return &Outbox{
49
49
logger: logger.With("component", "outbox"),
50
50
mode: parseOutboxMode(config.WebhookURL, config.DisableAcks),
+2
-2
cmd/tap/outbox_test.go
cmd/tab/outbox_test.go
+2
-2
cmd/tap/outbox_test.go
cmd/tab/outbox_test.go
···
4
4
"testing"
5
5
"time"
6
6
7
-
"github.com/bluesky-social/indigo/cmd/tap/models"
8
7
"github.com/stretchr/testify/assert"
9
8
"github.com/stretchr/testify/require"
9
+
"tangled.org/pds.dad/tab/cmd/tab/models"
10
10
)
11
11
12
12
func TestFireAndForget_BasicDelivery(t *testing.T) {
···
208
208
// TestWebsocketAck_FastPathLiveBlocksHistorical verifies that when a live event
209
209
// is sent via the DIDWorker's fast path (first event for a DID), subsequent
210
210
// historical events are blocked until the live event is acked.
211
-
// Regression test for: bugfix(cmd/tap): prevent DID worker live-barrier stall
211
+
// Regression test for: bugfix(cmd/tab): prevent DID worker live-barrier stall
212
212
func TestWebsocketAck_FastPathLiveBlocksHistorical(t *testing.T) {
213
213
te := newTestEnv(t, testEnvOpts{
214
214
outboxMode: OutboxModeWebsocketAck,
+1
-1
cmd/tap/repo_manager.go
cmd/tab/repo_manager.go
+1
-1
cmd/tap/repo_manager.go
cmd/tab/repo_manager.go
+2
-2
cmd/tap/resyncer.go
cmd/tab/resyncer.go
+2
-2
cmd/tap/resyncer.go
cmd/tab/resyncer.go
···
15
15
"github.com/bluesky-social/indigo/atproto/atdata"
16
16
repolib "github.com/bluesky-social/indigo/atproto/repo"
17
17
"github.com/bluesky-social/indigo/atproto/syntax"
18
-
"github.com/bluesky-social/indigo/cmd/tap/models"
19
18
"github.com/ipfs/go-cid"
20
19
"go.opentelemetry.io/otel/attribute"
21
20
"gorm.io/gorm"
21
+
"tangled.org/pds.dad/tab/cmd/tab/models"
22
22
)
23
23
24
24
type Resyncer struct {
···
38
38
pdsBackoffMu sync.RWMutex
39
39
}
40
40
41
-
func NewResyncer(logger *slog.Logger, db *gorm.DB, repos *RepoManager, events *EventManager, config *TapConfig) *Resyncer {
41
+
func NewResyncer(logger *slog.Logger, db *gorm.DB, repos *RepoManager, events *EventManager, config *tabConfig) *Resyncer {
42
42
return &Resyncer{
43
43
logger: logger.With("component", "resyncer"),
44
44
db: db,
+2
-2
cmd/tap/resyncer_test.go
cmd/tab/resyncer_test.go
+2
-2
cmd/tap/resyncer_test.go
cmd/tab/resyncer_test.go
···
6
6
"testing"
7
7
"time"
8
8
9
-
"github.com/bluesky-social/indigo/cmd/tap/models"
9
+
"tangled.org/pds.dad/tab/cmd/tab/models"
10
10
)
11
11
12
12
func newTestResyncer(te *testEnv) *Resyncer {
13
-
config := &TapConfig{
13
+
config := &tabConfig{
14
14
ResyncParallelism: 1,
15
15
RepoFetchTimeout: 30 * time.Second,
16
16
EventCacheSize: 1000,
+17
-17
cmd/tap/server.go
cmd/tab/server.go
+17
-17
cmd/tap/server.go
cmd/tab/server.go
···
8
8
"github.com/bluesky-social/indigo/atproto/auth"
9
9
"github.com/bluesky-social/indigo/atproto/identity"
10
10
"github.com/bluesky-social/indigo/atproto/syntax"
11
-
"github.com/bluesky-social/indigo/cmd/tap/models"
12
11
"github.com/gorilla/websocket"
13
12
"github.com/labstack/echo/v4"
14
13
"github.com/labstack/echo/v4/middleware"
15
14
"gorm.io/gorm"
16
15
"gorm.io/gorm/clause"
16
+
"tangled.org/pds.dad/tab/cmd/tab/models"
17
17
)
18
18
19
-
type TapServer struct {
19
+
type tabServer struct {
20
20
db *gorm.DB
21
21
echo *echo.Echo
22
22
logger *slog.Logger
···
28
28
crawler *Crawler
29
29
}
30
30
31
-
func NewTapServer(logger *slog.Logger, db *gorm.DB, outbox *Outbox, idDir identity.Directory, firehose *FirehoseProcessor, jetstream *JetstreamProcessor, crawler *Crawler, config *TapConfig) *TapServer {
32
-
return &TapServer{
31
+
func NewtabServer(logger *slog.Logger, db *gorm.DB, outbox *Outbox, idDir identity.Directory, firehose *FirehoseProcessor, jetstream *JetstreamProcessor, crawler *Crawler, config *tabConfig) *tabServer {
32
+
return &tabServer{
33
33
logger: logger.With("component", "server"),
34
34
db: db,
35
35
outbox: outbox,
···
41
41
}
42
42
}
43
43
44
-
func (ts *TapServer) Start(address string) error {
44
+
func (ts *tabServer) Start(address string) error {
45
45
ts.echo = echo.New()
46
46
ts.echo.HideBanner = true
47
47
ts.echo.HidePort = true // silence http server started on [::]:port log line
···
69
69
}
70
70
71
71
// Shutdown gracefully shuts down the HTTP server.
72
-
func (ts *TapServer) Shutdown(ctx context.Context) error {
72
+
func (ts *tabServer) Shutdown(ctx context.Context) error {
73
73
return ts.echo.Shutdown(ctx)
74
74
}
75
75
76
-
func (ts *TapServer) handleHealthcheck(c echo.Context) error {
76
+
func (ts *tabServer) handleHealthcheck(c echo.Context) error {
77
77
return c.JSON(http.StatusOK, map[string]string{
78
78
"status": "ok",
79
79
})
···
85
85
},
86
86
}
87
87
88
-
func (ts *TapServer) handleChannelWebsocket(c echo.Context) error {
88
+
func (ts *tabServer) handleChannelWebsocket(c echo.Context) error {
89
89
if ts.outbox.mode == OutboxModeWebhook {
90
90
return echo.NewHTTPError(http.StatusBadRequest, "websocket not available in webhook mode")
91
91
}
···
141
141
DIDs []string `json:"dids"`
142
142
}
143
143
144
-
func (ts *TapServer) handleAddRepos(c echo.Context) error {
144
+
func (ts *tabServer) handleAddRepos(c echo.Context) error {
145
145
ctx := c.Request().Context()
146
146
147
147
var payload DidPayload
···
167
167
return c.NoContent(http.StatusOK)
168
168
}
169
169
170
-
func (ts *TapServer) handleRemoveRepos(c echo.Context) error {
170
+
func (ts *tabServer) handleRemoveRepos(c echo.Context) error {
171
171
ctx := c.Request().Context()
172
172
173
173
var payload DidPayload
···
188
188
return c.NoContent(http.StatusOK)
189
189
}
190
190
191
-
func (ts *TapServer) handleResolveDID(c echo.Context) error {
191
+
func (ts *tabServer) handleResolveDID(c echo.Context) error {
192
192
didParam := c.Param("did")
193
193
194
194
did, err := syntax.ParseDID(didParam)
···
207
207
return c.JSON(http.StatusOK, ident.DIDDocument())
208
208
}
209
209
210
-
func (ts *TapServer) handleInfoRepo(c echo.Context) error {
210
+
func (ts *tabServer) handleInfoRepo(c echo.Context) error {
211
211
ctx := c.Request().Context()
212
212
did := c.Param("did")
213
213
···
233
233
})
234
234
}
235
235
236
-
func (ts *TapServer) handleStatsRepoCount(c echo.Context) error {
236
+
func (ts *tabServer) handleStatsRepoCount(c echo.Context) error {
237
237
ctx := c.Request().Context()
238
238
var count int64
239
239
if err := ts.db.WithContext(ctx).Model(&models.Repo{}).Count(&count).Error; err != nil {
···
242
242
return c.JSON(http.StatusOK, map[string]int64{"repo_count": count})
243
243
}
244
244
245
-
func (ts *TapServer) handleStatsRecordCount(c echo.Context) error {
245
+
func (ts *tabServer) handleStatsRecordCount(c echo.Context) error {
246
246
ctx := c.Request().Context()
247
247
var count int64
248
248
if err := ts.db.WithContext(ctx).Model(&models.RepoRecord{}).Count(&count).Error; err != nil {
···
251
251
return c.JSON(http.StatusOK, map[string]int64{"record_count": count})
252
252
}
253
253
254
-
func (ts *TapServer) handleStatsOutboxBuffer(c echo.Context) error {
254
+
func (ts *tabServer) handleStatsOutboxBuffer(c echo.Context) error {
255
255
ctx := c.Request().Context()
256
256
var count int64
257
257
if err := ts.db.WithContext(ctx).Model(&models.OutboxBuffer{}).Count(&count).Error; err != nil {
···
260
260
return c.JSON(http.StatusOK, map[string]int64{"outbox_buffer": count})
261
261
}
262
262
263
-
func (ts *TapServer) handleStatsResyncBuffer(c echo.Context) error {
263
+
func (ts *tabServer) handleStatsResyncBuffer(c echo.Context) error {
264
264
ctx := c.Request().Context()
265
265
var count int64
266
266
if err := ts.db.WithContext(ctx).Model(&models.ResyncBuffer{}).Count(&count).Error; err != nil {
···
275
275
ListRepos *string `json:"list_repos,omitempty"`
276
276
}
277
277
278
-
func (ts *TapServer) handleStatsCursors(c echo.Context) error {
278
+
func (ts *tabServer) handleStatsCursors(c echo.Context) error {
279
279
ctx := c.Request().Context()
280
280
resp := CursorsResp{}
281
281
+1
-1
cmd/tap/server_test.go
cmd/tab/server_test.go
+1
-1
cmd/tap/server_test.go
cmd/tab/server_test.go
+12
-12
cmd/tap/tap.go
cmd/tab/tap.go
+12
-12
cmd/tap/tap.go
cmd/tab/tap.go
···
8
8
"time"
9
9
10
10
"github.com/bluesky-social/indigo/atproto/identity"
11
-
"github.com/bluesky-social/indigo/cmd/tap/models"
12
11
"gorm.io/driver/postgres"
13
12
"gorm.io/driver/sqlite"
14
13
"gorm.io/gorm"
15
14
"gorm.io/gorm/logger"
15
+
"tangled.org/pds.dad/tab/cmd/tab/models"
16
16
)
17
17
18
-
type Tap struct {
18
+
type tab struct {
19
19
db *gorm.DB
20
20
logger *slog.Logger
21
21
···
26
26
resyncer *Resyncer
27
27
crawler *Crawler
28
28
29
-
server *TapServer
29
+
server *tabServer
30
30
outbox *Outbox
31
31
32
32
outboxOnly bool
33
33
}
34
34
35
-
type TapConfig struct {
35
+
type tabConfig struct {
36
36
DatabaseURL string
37
37
DBMaxConns int
38
38
PLCURL string
···
58
58
RetryTimeout time.Duration
59
59
}
60
60
61
-
func NewTap(config TapConfig) (*Tap, error) {
61
+
func Newtab(config tabConfig) (*tab, error) {
62
62
db, err := SetupDatabase(config.DatabaseURL, config.DBMaxConns)
63
63
if err != nil {
64
64
return nil, err
···
71
71
}
72
72
cdir := identity.NewCacheDirectory(&bdir, config.IdentityCacheSize, time.Hour*24, time.Minute*2, time.Minute*5)
73
73
74
-
logger := slog.Default().With("system", "tap")
74
+
logger := slog.Default().With("system", "tab")
75
75
76
76
evtMngr := NewEventManager(logger, db, &config)
77
77
···
91
91
92
92
outbox := NewOutbox(logger, evtMngr, &config)
93
93
94
-
server := NewTapServer(logger, db, outbox, cdir, firehose, jetstream, crawler, &config)
94
+
server := NewtabServer(logger, db, outbox, cdir, firehose, jetstream, crawler, &config)
95
95
96
-
t := &Tap{
96
+
t := &tab{
97
97
db: db,
98
-
logger: slog.Default().With("system", "tap"),
98
+
logger: slog.Default().With("system", "tab"),
99
99
100
100
firehose: firehose,
101
101
jetstream: jetstream,
···
116
116
}
117
117
118
118
// Run starts internal background workers for resync, cursor saving, and outbox delivery.
119
-
func (t *Tap) Run(ctx context.Context) {
119
+
func (t *tab) Run(ctx context.Context) {
120
120
go t.events.LoadEvents(ctx)
121
121
122
122
if !t.outboxOnly {
···
126
126
go t.outbox.Run(ctx)
127
127
}
128
128
129
-
func (t *Tap) CloseDb(ctx context.Context) error {
130
-
t.logger.Info("shutting down tap")
129
+
func (t *tab) CloseDb(ctx context.Context) error {
130
+
t.logger.Info("shutting down tab")
131
131
132
132
sqlDB, err := t.db.DB()
133
133
if err != nil {
+4
-4
cmd/tap/tap_test_helpers_test.go
cmd/tab/tap_test_helpers_test.go
+4
-4
cmd/tap/tap_test_helpers_test.go
cmd/tab/tap_test_helpers_test.go
···
14
14
"time"
15
15
16
16
"github.com/bluesky-social/indigo/atproto/identity"
17
-
"github.com/bluesky-social/indigo/cmd/tap/models"
18
17
"github.com/gorilla/websocket"
19
18
"gorm.io/gorm"
19
+
"tangled.org/pds.dad/tab/cmd/tab/models"
20
20
)
21
21
22
22
type testEnvOpts struct {
···
33
33
db *gorm.DB
34
34
events *EventManager
35
35
outbox *Outbox
36
-
server *TapServer
36
+
server *tabServer
37
37
repos *RepoManager
38
38
idDir *identity.MockDirectory
39
39
port int
···
74
74
disableAcks = true
75
75
}
76
76
77
-
config := &TapConfig{
77
+
config := &tabConfig{
78
78
EventCacheSize: cacheSize,
79
79
OutboxParallelism: 1,
80
80
DisableAcks: disableAcks,
···
88
88
events := NewEventManager(logger, db, config)
89
89
repos := NewRepoManager(logger, db, events, idDir)
90
90
outbox := NewOutbox(logger, events, config)
91
-
server := NewTapServer(logger, db, outbox, idDir, nil, nil, nil, config)
91
+
server := NewtabServer(logger, db, outbox, idDir, nil, nil, nil, config)
92
92
93
93
// Find a free port
94
94
listener, err := net.Listen("tcp", "127.0.0.1:0")
+1
-1
cmd/tap/types.go
cmd/tab/types.go
+1
-1
cmd/tap/types.go
cmd/tab/types.go
+1
-1
cmd/tap/util.go
cmd/tab/util.go
+1
-1
cmd/tap/util.go
cmd/tab/util.go
cmd/tap/util_test.go
cmd/tab/util_test.go
cmd/tap/util_test.go
cmd/tab/util_test.go
cmd/tap/webhook_client.go
cmd/tab/webhook_client.go
cmd/tap/webhook_client.go
cmd/tab/webhook_client.go