+3

tapc/readme.md

··· 1 1 + basic tap client package 2 2 + 3 3 + Replace this to official indigo package when <https://github.com/bluesky-social/indigo/pull/1241> gets merged.

+24

tapc/simpleIndexer.go

··· 1 1 + package tapc 2 2 + 3 3 + import "context" 4 4 + 5 5 + type SimpleIndexer struct { 6 6 + EventHandler func(ctx context.Context, evt Event) error 7 7 + ErrorHandler func(ctx context.Context, err error) 8 8 + } 9 9 + 10 10 + var _ Handler = (*SimpleIndexer)(nil) 11 11 + 12 12 + func (i *SimpleIndexer) OnEvent(ctx context.Context, evt Event) error { 13 13 + if i.EventHandler == nil { 14 14 + return nil 15 15 + } 16 16 + return i.EventHandler(ctx, evt) 17 17 + } 18 18 + 19 19 + func (i *SimpleIndexer) OnError(ctx context.Context, err error) { 20 20 + if i.ErrorHandler == nil { 21 21 + return 22 22 + } 23 23 + i.ErrorHandler(ctx, err) 24 24 + }

+176

tapc/tap.go

··· 1 1 + /// heavily inspired by <https://github.com/bluesky-social/atproto/blob/c7f5a868837d3e9b3289f988fee2267789327b06/packages/tap/README.md> 2 2 + 3 3 + package tapc 4 4 + 5 5 + import ( 6 6 + "bytes" 7 7 + "context" 8 8 + "encoding/json" 9 9 + "fmt" 10 10 + "net/http" 11 11 + "net/url" 12 12 + "time" 13 13 + 14 14 + "github.com/bluesky-social/indigo/atproto/syntax" 15 15 + "github.com/gorilla/websocket" 16 16 + "tangled.org/core/log" 17 17 + ) 18 18 + 19 19 + // type WebsocketOptions struct { 20 20 + // maxReconnectSeconds int 21 21 + // heartbeatIntervalMs int 22 22 + // // onReconnectError 23 23 + // } 24 24 + 25 25 + type Handler interface { 26 26 + OnEvent(ctx context.Context, evt Event) error 27 27 + OnError(ctx context.Context, err error) 28 28 + } 29 29 + 30 30 + type Client struct { 31 31 + Url string 32 32 + AdminPassword string 33 33 + HTTPClient *http.Client 34 34 + } 35 35 + 36 36 + func NewClient(url, adminPassword string) Client { 37 37 + return Client{ 38 38 + Url: url, 39 39 + AdminPassword: adminPassword, 40 40 + HTTPClient: &http.Client{}, 41 41 + } 42 42 + } 43 43 + 44 44 + func (c *Client) AddRepos(ctx context.Context, dids []syntax.DID) error { 45 45 + body, err := json.Marshal(map[string][]syntax.DID{"dids": dids}) 46 46 + if err != nil { 47 47 + return err 48 48 + } 49 49 + req, err := http.NewRequestWithContext(ctx, "POST", c.Url+"/repos/add", bytes.NewReader(body)) 50 50 + if err != nil { 51 51 + return err 52 52 + } 53 53 + req.SetBasicAuth("admin", c.AdminPassword) 54 54 + req.Header.Set("Content-Type", "application/json") 55 55 + 56 56 + resp, err := c.HTTPClient.Do(req) 57 57 + if err != nil { 58 58 + return err 59 59 + } 60 60 + defer resp.Body.Close() 61 61 + if resp.StatusCode != http.StatusOK { 62 62 + return fmt.Errorf("tap: /repos/add failed with status %d", resp.StatusCode) 63 63 + } 64 64 + return nil 65 65 + } 66 66 + 67 67 + func (c *Client) RemoveRepos(ctx context.Context, dids []syntax.DID) error { 68 68 + body, err := json.Marshal(map[string][]syntax.DID{"dids": dids}) 69 69 + if err != nil { 70 70 + return err 71 71 + } 72 72 + req, err := http.NewRequestWithContext(ctx, "POST", c.Url+"/repos/remove", bytes.NewReader(body)) 73 73 + if err != nil { 74 74 + return err 75 75 + } 76 76 + req.SetBasicAuth("admin", c.AdminPassword) 77 77 + req.Header.Set("Content-Type", "application/json") 78 78 + 79 79 + resp, err := c.HTTPClient.Do(req) 80 80 + if err != nil { 81 81 + return err 82 82 + } 83 83 + defer resp.Body.Close() 84 84 + if resp.StatusCode != http.StatusOK { 85 85 + return fmt.Errorf("tap: /repos/remove failed with status %d", resp.StatusCode) 86 86 + } 87 87 + return nil 88 88 + } 89 89 + 90 90 + func (c *Client) Connect(ctx context.Context, handler Handler) error { 91 91 + l := log.FromContext(ctx) 92 92 + 93 93 + u, err := url.Parse(c.Url) 94 94 + if err != nil { 95 95 + return err 96 96 + } 97 97 + if u.Scheme == "https" { 98 98 + u.Scheme = "wss" 99 99 + } else { 100 100 + u.Scheme = "ws" 101 101 + } 102 102 + u.Path = "/channel" 103 103 + 104 104 + // TODO: set auth on dial 105 105 + 106 106 + url := u.String() 107 107 + 108 108 + var backoff int 109 109 + for { 110 110 + select { 111 111 + case <-ctx.Done(): 112 112 + return ctx.Err() 113 113 + default: 114 114 + } 115 115 + 116 116 + header := http.Header{ 117 117 + "Authorization": []string{""}, 118 118 + } 119 119 + conn, res, err := websocket.DefaultDialer.DialContext(ctx, url, header) 120 120 + if err != nil { 121 121 + l.Warn("dialing failed", "url", url, "err", err, "backoff", backoff) 122 122 + time.Sleep(time.Duration(5+backoff) * time.Second) 123 123 + backoff++ 124 124 + 125 125 + continue 126 126 + } 127 127 + l.Info("connected to tap service") 128 128 + 129 129 + l.Info("tap event subscription response", "code", res.StatusCode) 130 130 + 131 131 + if err = c.handleConnection(ctx, conn, handler); err != nil { 132 132 + l.Warn("tap connection failed", "err", err, "backoff", backoff) 133 133 + } 134 134 + } 135 135 + } 136 136 + 137 137 + func (c *Client) handleConnection(ctx context.Context, conn *websocket.Conn, handler Handler) error { 138 138 + l := log.FromContext(ctx) 139 139 + 140 140 + defer func() { 141 141 + conn.Close() 142 142 + l.Warn("closed tap conection") 143 143 + }() 144 144 + l.Info("established tap conection") 145 145 + 146 146 + for { 147 147 + select { 148 148 + case <-ctx.Done(): 149 149 + return ctx.Err() 150 150 + default: 151 151 + } 152 152 + _, message, err := conn.ReadMessage() 153 153 + if err != nil { 154 154 + return err 155 155 + } 156 156 + 157 157 + var ev Event 158 158 + if err := json.Unmarshal(message, &ev); err != nil { 159 159 + handler.OnError(ctx, fmt.Errorf("failed to parse message: %w", err)) 160 160 + continue 161 161 + } 162 162 + if err := handler.OnEvent(ctx, ev); err != nil { 163 163 + handler.OnError(ctx, fmt.Errorf("failed to process event %d: %w", ev.ID, err)) 164 164 + continue 165 165 + } 166 166 + 167 167 + ack := map[string]any{ 168 168 + "type": "ack", 169 169 + "id": ev.ID, 170 170 + } 171 171 + if err := conn.WriteJSON(ack); err != nil { 172 172 + l.Warn("failed to send ack", "err", err) 173 173 + continue 174 174 + } 175 175 + } 176 176 + }

+62

tapc/types.go

··· 1 1 + package tapc 2 2 + 3 3 + import ( 4 4 + "encoding/json" 5 5 + "fmt" 6 6 + 7 7 + "github.com/bluesky-social/indigo/atproto/syntax" 8 8 + ) 9 9 + 10 10 + type EventType string 11 11 + 12 12 + const ( 13 13 + EvtRecord EventType = "record" 14 14 + EvtIdentity EventType = "identity" 15 15 + ) 16 16 + 17 17 + type Event struct { 18 18 + ID int64 `json:"id"` 19 19 + Type EventType `json:"type"` 20 20 + Record *RecordEventData `json:"record,omitempty"` 21 21 + Identity *IdentityEventData `json:"identity,omitempty"` 22 22 + } 23 23 + 24 24 + type RecordEventData struct { 25 25 + Live bool `json:"live"` 26 26 + Did syntax.DID `json:"did"` 27 27 + Rev string `json:"rev"` 28 28 + Collection syntax.NSID `json:"collection"` 29 29 + Rkey syntax.RecordKey `json:"rkey"` 30 30 + Action RecordAction `json:"action"` 31 31 + Record json.RawMessage `json:"record,omitempty"` 32 32 + CID *syntax.CID `json:"cid,omitempty"` 33 33 + } 34 34 + 35 35 + func (r *RecordEventData) AtUri() syntax.ATURI { 36 36 + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", r.Did, r.Collection, r.Rkey)) 37 37 + } 38 38 + 39 39 + type RecordAction string 40 40 + 41 41 + const ( 42 42 + RecordCreateAction RecordAction = "create" 43 43 + RecordUpdateAction RecordAction = "update" 44 44 + RecordDeleteAction RecordAction = "delete" 45 45 + ) 46 46 + 47 47 + type IdentityEventData struct { 48 48 + DID syntax.DID `json:"did"` 49 49 + Handle string `json:"handle"` 50 50 + IsActive bool `json:"is_active"` 51 51 + Status RepoStatus `json:"status"` 52 52 + } 53 53 + 54 54 + type RepoStatus string 55 55 + 56 56 + const ( 57 57 + RepoStatusActive RepoStatus = "active" 58 58 + RepoStatusTakendown RepoStatus = "takendown" 59 59 + RepoStatusSuspended RepoStatus = "suspended" 60 60 + RepoStatusDeactivated RepoStatus = "deactivated" 61 61 + RepoStatusDeleted RepoStatus = "deleted" 62 62 + )