+24

tap/simpleIndexer.go

··· 1 1 + package tap 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 + }

+169

tap/tap.go

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

+62

tap/types.go

··· 1 1 + package tap 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 + )