diff --git a/tapc/readme.md b/tapc/readme.md new file mode 100644 --- /dev/null +++ b/tapc/readme.md @@ -0,0 +1,3 @@ +basic tap client package + +Replace this to official indigo package when gets merged. diff --git a/tapc/simpleIndexer.go b/tapc/simpleIndexer.go new file mode 100644 --- /dev/null +++ b/tapc/simpleIndexer.go @@ -0,0 +1,24 @@ +package tapc + +import "context" + +type SimpleIndexer struct { + EventHandler func(ctx context.Context, evt Event) error + ErrorHandler func(ctx context.Context, err error) +} + +var _ Handler = (*SimpleIndexer)(nil) + +func (i *SimpleIndexer) OnEvent(ctx context.Context, evt Event) error { + if i.EventHandler == nil { + return nil + } + return i.EventHandler(ctx, evt) +} + +func (i *SimpleIndexer) OnError(ctx context.Context, err error) { + if i.ErrorHandler == nil { + return + } + i.ErrorHandler(ctx, err) +} diff --git a/tapc/tap.go b/tapc/tap.go new file mode 100644 --- /dev/null +++ b/tapc/tap.go @@ -0,0 +1,170 @@ +/// heavily inspired by + +package tapc + +import ( + "bytes" + "context" + "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 Client struct { + Url string + AdminPassword string + HTTPClient *http.Client +} + +func NewClient(url, adminPassword string) Client { + return Client{ + Url: url, + AdminPassword: adminPassword, + HTTPClient: &http.Client{}, + } +} + +func (c *Client) AddRepos(ctx context.Context, 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+"/repos/add", 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: /repos/add failed with status %d", resp.StatusCode) + } + return nil +} + +func (c *Client) RemoveRepos(ctx context.Context, 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+"/repos/remove", 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: /repos/remove failed with status %d", 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" + + // TODO: set auth on dial + + url := u.String() + + var backoff int + for { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + header := http.Header{ + "Authorization": []string{""}, + } + conn, res, err := websocket.DefaultDialer.DialContext(ctx, url, header) + if err != nil { + l.Warn("dialing failed", "url", url, "err", err, "backoff", backoff) + time.Sleep(time.Duration(5+backoff) * time.Second) + backoff++ + + continue + } + l.Info("connected to tap service") + + l.Info("tap event subscription response", "code", res.StatusCode) + + if err = c.handleConnection(ctx, conn, handler); err != nil { + l.Warn("tap connection failed", "err", err, "backoff", backoff) + } + } +} + +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 conection") + }() + l.Info("established tap conection") + + 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 + } + } +} diff --git a/tapc/types.go b/tapc/types.go new file mode 100644 --- /dev/null +++ b/tapc/types.go @@ -0,0 +1,62 @@ +package tapc + +import ( + "encoding/json" + "fmt" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +type EventType string + +const ( + EvtRecord EventType = "record" + EvtIdentity EventType = "identity" +) + +type Event struct { + ID int64 `json:"id"` + Type EventType `json:"type"` + Record *RecordEventData `json:"record,omitempty"` + Identity *IdentityEventData `json:"identity,omitempty"` +} + +type RecordEventData struct { + Live bool `json:"live"` + Did syntax.DID `json:"did"` + Rev string `json:"rev"` + Collection syntax.NSID `json:"collection"` + Rkey syntax.RecordKey `json:"rkey"` + Action RecordAction `json:"action"` + Record json.RawMessage `json:"record,omitempty"` + CID *syntax.CID `json:"cid,omitempty"` +} + +func (r *RecordEventData) AtUri() syntax.ATURI { + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", r.Did, r.Collection, r.Rkey)) +} + +type RecordAction string + +const ( + RecordCreateAction RecordAction = "create" + RecordUpdateAction RecordAction = "update" + RecordDeleteAction RecordAction = "delete" +) + +type IdentityEventData struct { + DID syntax.DID `json:"did"` + Handle string `json:"handle"` + IsActive bool `json:"is_active"` + Status RepoStatus `json:"status"` +} + +type RepoStatus string + +const ( + RepoStatusActive RepoStatus = "active" + RepoStatusTakendown RepoStatus = "takendown" + RepoStatusSuspended RepoStatus = "suspended" + RepoStatusDeactivated RepoStatus = "deactivated" + RepoStatusDeleted RepoStatus = "deleted" +)