tap: add tap client package · lewis.moe/tangled-core@e6f933a

+24

tap/simpleIndexer.go

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

+169

tap/tap.go

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

+62

tap/types.go

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