/// heavily inspired by <https://github.com/bluesky-social/atproto/blob/c7f5a868837d3e9b3289f988fee2267789327b06/packages/tap/README.md>
package tapc
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"net/url"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/gorilla/websocket"
"tangled.org/core/log"
)
type Handler interface {
OnEvent(ctx context.Context, evt Event) error
OnError(ctx context.Context, err error)
}
type ConnectHandler interface {
OnConnect(ctx context.Context)
}
type Client struct {
Url string
AdminPassword string
HTTPClient *http.Client
}
func NewClient(url, adminPassword string) Client {
return Client{
Url: url,
AdminPassword: adminPassword,
HTTPClient: &http.Client{},
}
}
// tap inserts one /repos/add body as a single sqlite statement at about 9
// variables per did, so a spindle's whole member list blows past sqlite's
// 32766 cap and tap 500s
const addReposBatchSize = 1000
func (c *Client) AddRepos(ctx context.Context, dids []syntax.DID) error {
for len(dids) > 0 {
n := min(len(dids), addReposBatchSize)
if err := c.postDids(ctx, "/repos/add", dids[:n]); err != nil {
return err
}
dids = dids[n:]
}
return nil
}
func (c *Client) RemoveRepos(ctx context.Context, dids []syntax.DID) error {
return c.postDids(ctx, "/repos/remove", dids)
}
func (c *Client) postDids(ctx context.Context, path string, dids []syntax.DID) error {
body, err := json.Marshal(map[string][]syntax.DID{"dids": dids})
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, "POST", c.Url+path, bytes.NewReader(body))
if err != nil {
return err
}
req.SetBasicAuth("admin", c.AdminPassword)
req.Header.Set("Content-Type", "application/json")
resp, err := c.HTTPClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("tap: %s failed with status %d", path, resp.StatusCode)
}
return nil
}
func (c *Client) Connect(ctx context.Context, handler Handler) error {
l := log.FromContext(ctx)
u, err := url.Parse(c.Url)
if err != nil {
return err
}
if u.Scheme == "https" {
u.Scheme = "wss"
} else {
u.Scheme = "ws"
}
u.Path = "/channel"
url := u.String()
basicAuth := "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+c.AdminPassword))
var backoff int
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
header := http.Header{
"Authorization": []string{basicAuth},
}
conn, res, err := websocket.DefaultDialer.DialContext(ctx, url, header)
if err != nil {
if backoff < 12 {
backoff++
}
l.Warn("dialing failed", "url", url, "err", err, "backoff", backoff)
time.Sleep(time.Duration(5*backoff) * time.Second)
continue
}
backoff = 0
l.Info("connected to tap service", "subscription_code", res.StatusCode)
if ch, ok := handler.(ConnectHandler); ok {
ch.OnConnect(ctx)
}
if err = c.handleConnection(ctx, conn, handler); err != nil {
l.Warn("tap connection failed", "err", err)
}
}
}
func (c *Client) handleConnection(ctx context.Context, conn *websocket.Conn, handler Handler) error {
l := log.FromContext(ctx)
defer func() {
conn.Close()
l.Warn("closed tap connection")
}()
l.Info("established tap connection")
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
_, message, err := conn.ReadMessage()
if err != nil {
return err
}
var ev Event
if err := json.Unmarshal(message, &ev); err != nil {
handler.OnError(ctx, fmt.Errorf("failed to parse message: %w", err))
continue
}
if err := handler.OnEvent(ctx, ev); err != nil {
handler.OnError(ctx, fmt.Errorf("failed to process event %d: %w", ev.ID, err))
continue
}
ack := map[string]any{
"type": "ack",
"id": ev.ID,
}
if err := conn.WriteJSON(ack); err != nil {
l.Warn("failed to send ack", "err", err)
continue
}
}
}