Rough draft for working out who to send the alert to and allowing users to subscribe via dm #3

+28

cmd/main.go

··· 12 12 "os/signal" 13 13 "path" 14 14 "syscall" 15 15 + "time" 15 16 16 17 tangledalertbot "tangled.sh/willdot.net/tangled-alert-bot" 17 18 ··· 56 57 } 57 58 defer database.Close() 58 59 60 60 + dmService, err := tangledalertbot.NewDmService(database, time.Second*30) 61 61 + if err != nil { 62 62 + return fmt.Errorf("create dm service: %w", err) 63 63 + } 64 64 + 59 65 ctx, cancel := context.WithCancel(context.Background()) 60 66 defer cancel() 61 67 ··· 63 69 64 70 go startHttpServer(ctx, database) 65 71 72 72 + go dmService.Start(ctx) 73 73 + 66 74 <-signals 67 75 cancel() 68 76 ··· 101 109 mux := http.NewServeMux() 102 110 mux.HandleFunc("/issues", srv.handleListIssues) 103 111 mux.HandleFunc("/comments", srv.handleListComments) 112 112 + mux.HandleFunc("/users", srv.handleListUsers) 104 113 105 114 err := http.ListenAndServe(":3000", mux) 106 115 if err != nil { ··· 149 158 w.Header().Set("Content-Type", "application/json") 150 159 w.Write(b) 151 160 } 161 161 + 162 162 + func (s *server) handleListUsers(w http.ResponseWriter, r *http.Request) { 163 163 + users, err := s.db.GetUsers() 164 164 + if err != nil { 165 165 + slog.Error("getting users from DB", "error", err) 166 166 + http.Error(w, "error getting users from DB", http.StatusInternalServerError) 167 167 + return 168 168 + } 169 169 + 170 170 + b, err := json.Marshal(users) 171 171 + if err != nil { 172 172 + slog.Error("marshalling users from DB", "error", err) 173 173 + http.Error(w, "marshalling users from DB", http.StatusInternalServerError) 174 174 + return 175 175 + } 176 176 + 177 177 + w.Header().Set("Content-Type", "application/json") 178 178 + w.Write(b) 179 179 + }

+49 -1

consumer.go

··· 3 3 import ( 4 4 "context" 5 5 "encoding/json" 6 6 + "strings" 6 7 7 8 "fmt" 8 9 "log/slog" ··· 38 39 DeleteIssue(did, rkey string) error 39 40 DeleteComment(did, rkey string) error 40 41 DeleteCommentsForIssue(issueURI string) error 42 42 + GetUser(did string) (User, error) 43 43 + CreateUser(user User) error 41 44 } 42 45 43 46 // JetstreamConsumer is responsible for consuming from a jetstream instance ··· 217 220 } 218 221 219 222 // TODO: now send a notification to either the issue creator or whoever the comment was a reply to 223 223 + didToNotify := getUserToAlert(comment) 224 224 + if didToNotify == "" { 225 225 + slog.Info("could not work out did to send alert to", "comment", comment) 226 226 + return 227 227 + } 228 228 + 229 229 + user, err := h.store.GetUser(didToNotify) 230 230 + if err != nil { 231 231 + slog.Error("getting user to send alert to", "error", err, "did", didToNotify) 232 232 + return 233 233 + } 220 234 221 221 - slog.Info("created comment ", "value", comment, "did", did, "rkey", rkey) 235 235 + slog.Info("sending alert to user", "value", comment, "did", didToNotify, "convo", user.ConvoID) 222 236 } 223 237 224 238 func (h *Handler) handleDeleteIssueEvent(ctx context.Context, event *models.Event) { ··· 259 273 260 274 slog.Info("deleted comment ", "did", did, "rkey", rkey) 261 275 } 276 276 + 277 277 + // at://did:plc:dadhhalkfcq3gucaq25hjqon/sh.tangled.repo.issue.comment/3lzkp4va62m22 278 278 + func getUserToAlert(comment tangled.RepoIssueComment) string { 279 279 + if comment.ReplyTo != nil { 280 280 + return getDidFromCommentURI(*comment.ReplyTo) 281 281 + } 282 282 + return getDidFromIssueURI(comment.Issue) 283 283 + } 284 284 + 285 285 + func getDidFromCommentURI(uri string) string { 286 286 + split := strings.Split(uri, tangled.RepoIssueCommentNSID) 287 287 + if len(split) != 2 { 288 288 + slog.Error("invalid comment URI received", "uri", uri) 289 289 + return "" 290 290 + } 291 291 + 292 292 + did := strings.TrimPrefix(split[0], "at://") 293 293 + did = strings.TrimSuffix(did, "/") 294 294 + 295 295 + return did 296 296 + } 297 297 + 298 298 + func getDidFromIssueURI(uri string) string { 299 299 + split := strings.Split(uri, tangled.RepoIssueNSID) 300 300 + if len(split) != 2 { 301 301 + slog.Error("invalid issue URI received", "uri", uri) 302 302 + return "" 303 303 + } 304 304 + 305 305 + did := strings.TrimPrefix(split[0], "at://") 306 306 + did = strings.TrimSuffix(did, "/") 307 307 + 308 308 + return did 309 309 + }

+86

database.go

··· 44 44 return nil, fmt.Errorf("creating comments table: %w", err) 45 45 } 46 46 47 47 + err = createUsersTable(db) 48 48 + if err != nil { 49 49 + return nil, fmt.Errorf("creating users table: %w", err) 50 50 + } 51 51 + 47 52 return &Database{db: db}, nil 48 53 } 49 54 ··· 122 127 return nil 123 128 } 124 129 130 130 + func createUsersTable(db *sql.DB) error { 131 131 + createTableSQL := `CREATE TABLE IF NOT EXISTS users ( 132 132 + "id" integer NOT NULL PRIMARY KEY AUTOINCREMENT, 133 133 + "did" TEXT, 134 134 + "handle" TEXT, 135 135 + "convoId" TEXT, 136 136 + "createdAt" integer NOT NULL, 137 137 + UNIQUE(did) 138 138 + );` 139 139 + 140 140 + slog.Info("Create users table...") 141 141 + statement, err := db.Prepare(createTableSQL) 142 142 + if err != nil { 143 143 + return fmt.Errorf("prepare DB statement to create users table: %w", err) 144 144 + } 145 145 + _, err = statement.Exec() 146 146 + if err != nil { 147 147 + return fmt.Errorf("exec sql statement to create users table: %w", err) 148 148 + } 149 149 + slog.Info("users table created") 150 150 + 151 151 + return nil 152 152 + } 153 153 + 125 154 // CreateIssue will insert a issue into a database 126 155 func (d *Database) CreateIssue(issue Issue) error { 127 156 sql := `REPLACE INTO issues (authorDid, rkey, title, body, repo, createdAt) VALUES (?, ?, ?, ?, ?, ?);` ··· 142 171 return nil 143 172 } 144 173 174 174 + // CreateUser will insert a user into a database 175 175 + func (d *Database) CreateUser(user User) error { 176 176 + sql := `REPLACE INTO users (did, handle, convoId, createdAt) VALUES (?, ?, ?, ?);` 177 177 + _, err := d.db.Exec(sql, user.DID, user.Handle, user.ConvoID, user.CreatedAt) 178 178 + if err != nil { 179 179 + return fmt.Errorf("exec insert user: %w", err) 180 180 + } 181 181 + return nil 182 182 + } 183 183 + 145 184 func (d *Database) GetIssues() ([]Issue, error) { 146 185 sql := "SELECT authorDid, rkey, title, body, repo, createdAt FROM issues;" 147 186 rows, err := d.db.Query(sql) ··· 182 221 return results, nil 183 222 } 184 223 224 224 + func (d *Database) GetUser(did string) (User, error) { 225 225 + sql := "SELECT did, handle, convoId, createdAt FROM users WHERE did = ?;" 226 226 + rows, err := d.db.Query(sql, did) 227 227 + if err != nil { 228 228 + return User{}, fmt.Errorf("run query to get user: %w", err) 229 229 + } 230 230 + defer rows.Close() 231 231 + 232 232 + for rows.Next() { 233 233 + var user User 234 234 + if err := rows.Scan(&user.DID, &user.Handle, &user.ConvoID, &user.CreatedAt); err != nil { 235 235 + return User{}, fmt.Errorf("scan row: %w", err) 236 236 + } 237 237 + 238 238 + return user, nil 239 239 + } 240 240 + return User{}, fmt.Errorf("user not found") 241 241 + } 242 242 + 243 243 + func (d *Database) GetUsers() ([]User, error) { 244 244 + sql := "SELECT did, handle, convoId, createdAt FROM users;" 245 245 + rows, err := d.db.Query(sql) 246 246 + if err != nil { 247 247 + return nil, fmt.Errorf("run query to get user: %w", err) 248 248 + } 249 249 + defer rows.Close() 250 250 + 251 251 + var results []User 252 252 + for rows.Next() { 253 253 + var user User 254 254 + if err := rows.Scan(&user.DID, &user.Handle, &user.ConvoID, &user.CreatedAt); err != nil { 255 255 + return nil, fmt.Errorf("scan row: %w", err) 256 256 + } 257 257 + results = append(results, user) 258 258 + } 259 259 + return results, nil 260 260 + } 261 261 + 185 262 func (d *Database) DeleteIssue(did, rkey string) error { 186 263 sql := "DELETE FROM issues WHERE authorDid = ? AND rkey = ?;" 187 264 _, err := d.db.Exec(sql, did, rkey) ··· 208 285 } 209 286 return nil 210 287 } 288 288 + 289 289 + func (d *Database) DeleteUser(did string) error { 290 290 + sql := "DELETE FROM users WHERE did = ?;" 291 291 + _, err := d.db.Exec(sql, did) 292 292 + if err != nil { 293 293 + return fmt.Errorf("exec delete user") 294 294 + } 295 295 + return nil 296 296 + }

+441

dm_handler.go

··· 1 1 + package tangledalertbot 2 2 + 3 3 + import ( 4 4 + "bytes" 5 5 + "context" 6 6 + "encoding/json" 7 7 + "fmt" 8 8 + "io" 9 9 + "log/slog" 10 10 + "net/http" 11 11 + "os" 12 12 + "strings" 13 13 + "time" 14 14 + 15 15 + "github.com/pkg/errors" 16 16 + ) 17 17 + 18 18 + const ( 19 19 + httpClientTimeoutDuration = time.Second * 5 20 20 + transportIdleConnTimeoutDuration = time.Second * 90 21 21 + baseBskyURL = "https://bsky.social/xrpc" 22 22 + ) 23 23 + 24 24 + type auth struct { 25 25 + AccessJwt string `json:"accessJwt"` 26 26 + RefershJWT string `json:"refreshJwt"` 27 27 + Did string `json:"did"` 28 28 + } 29 29 + 30 30 + type accessData struct { 31 31 + handle string 32 32 + appPassword string 33 33 + } 34 34 + 35 35 + type ListConvosResponse struct { 36 36 + Cursor string `json:"cursor"` 37 37 + Convos []Convo `json:"convos"` 38 38 + } 39 39 + 40 40 + type Convo struct { 41 41 + ID string `json:"id"` 42 42 + Members []ConvoMember `json:"members"` 43 43 + UnreadCount int `json:"unreadCount"` 44 44 + } 45 45 + 46 46 + type ConvoMember struct { 47 47 + Did string `json:"did"` 48 48 + Handle string `json:"handle"` 49 49 + } 50 50 + 51 51 + type ErrorResponse struct { 52 52 + Error string `json:"error"` 53 53 + } 54 54 + 55 55 + type MessageResp struct { 56 56 + Messages []Message `json:"messages"` 57 57 + Cursor string `json:"cursor"` 58 58 + } 59 59 + 60 60 + type Message struct { 61 61 + ID string `json:"id"` 62 62 + Sender MessageSender `json:"sender"` 63 63 + Text string `json:"text"` 64 64 + } 65 65 + 66 66 + type MessageSender struct { 67 67 + Did string `json:"did"` 68 68 + } 69 69 + 70 70 + type UpdateMessageReadRequest struct { 71 71 + ConvoID string `json:"convoId"` 72 72 + MessageID string `json:"messageId"` 73 73 + } 74 74 + 75 75 + type User struct { 76 76 + DID string 77 77 + Handle string 78 78 + ConvoID string 79 79 + CreatedAt int 80 80 + } 81 81 + 82 82 + type DmService struct { 83 83 + httpClient *http.Client 84 84 + accessData accessData 85 85 + auth auth 86 86 + timerDuration time.Duration 87 87 + pdsURL string 88 88 + store Store 89 89 + } 90 90 + 91 91 + func NewDmService(store Store, timerDuration time.Duration) (*DmService, error) { 92 92 + httpClient := http.Client{ 93 93 + Timeout: httpClientTimeoutDuration, 94 94 + Transport: &http.Transport{ 95 95 + IdleConnTimeout: transportIdleConnTimeoutDuration, 96 96 + }, 97 97 + } 98 98 + 99 99 + accessHandle := os.Getenv("MESSAGING_ACCESS_HANDLE") 100 100 + accessAppPassword := os.Getenv("MESSAGING_ACCESS_APP_PASSWORD") 101 101 + pdsURL := os.Getenv("MESSAGING_PDS_URL") 102 102 + 103 103 + service := DmService{ 104 104 + httpClient: &httpClient, 105 105 + accessData: accessData{ 106 106 + handle: accessHandle, 107 107 + appPassword: accessAppPassword, 108 108 + }, 109 109 + timerDuration: timerDuration, 110 110 + pdsURL: pdsURL, 111 111 + store: store, 112 112 + } 113 113 + 114 114 + auth, err := service.Authenicate() 115 115 + if err != nil { 116 116 + return nil, fmt.Errorf("authenticating: %w", err) 117 117 + } 118 118 + 119 119 + service.auth = auth 120 120 + 121 121 + return &service, nil 122 122 + } 123 123 + 124 124 + func (d *DmService) Start(ctx context.Context) { 125 125 + go d.RefreshTask(ctx) 126 126 + 127 127 + timer := time.NewTimer(d.timerDuration) 128 128 + defer timer.Stop() 129 129 + 130 130 + for { 131 131 + select { 132 132 + case <-ctx.Done(): 133 133 + slog.Warn("context canceled - stopping dm task") 134 134 + return 135 135 + case <-timer.C: 136 136 + err := d.HandleMessageTimer(ctx) 137 137 + if err != nil { 138 138 + slog.Error("handle message timer", "error", err) 139 139 + } 140 140 + timer.Reset(d.timerDuration) 141 141 + } 142 142 + } 143 143 + } 144 144 + 145 145 + func (d *DmService) RefreshTask(ctx context.Context) { 146 146 + timer := time.NewTimer(time.Hour) 147 147 + defer timer.Stop() 148 148 + 149 149 + for { 150 150 + select { 151 151 + case <-ctx.Done(): 152 152 + return 153 153 + case <-timer.C: 154 154 + err := d.RefreshAuthenication(ctx) 155 155 + if err != nil { 156 156 + slog.Error("handle refresh auth timer", "error", err) 157 157 + // TODO: better retry with backoff probably 158 158 + timer.Reset(time.Minute) 159 159 + continue 160 160 + } 161 161 + timer.Reset(time.Hour) 162 162 + } 163 163 + } 164 164 + } 165 165 + 166 166 + func (d *DmService) HandleMessageTimer(ctx context.Context) error { 167 167 + convoResp, err := d.GetUnreadMessages() 168 168 + if err != nil { 169 169 + return fmt.Errorf("get unread messages: %w", err) 170 170 + } 171 171 + 172 172 + // TODO: handle the cursor pagination 173 173 + 174 174 + for _, convo := range convoResp.Convos { 175 175 + if convo.UnreadCount == 0 { 176 176 + continue 177 177 + } 178 178 + 179 179 + messageResp, err := d.GetMessages(ctx, convo.ID) 180 180 + if err != nil { 181 181 + slog.Error("failed to get messages for convo", "error", err, "convo id", convo.ID) 182 182 + continue 183 183 + } 184 184 + 185 185 + unreadCount := convo.UnreadCount 186 186 + unreadMessages := make([]Message, 0, convo.UnreadCount) 187 187 + // TODO: handle cursor pagination 188 188 + for _, msg := range messageResp.Messages { 189 189 + // TODO: techincally if I get to a message that's from the bot account, then there shouldn't be 190 190 + // an more unread messages? 191 191 + if msg.Sender.Did == d.auth.Did { 192 192 + continue 193 193 + } 194 194 + 195 195 + unreadMessages = append(unreadMessages, msg) 196 196 + unreadCount-- 197 197 + if unreadCount == 0 { 198 198 + break 199 199 + } 200 200 + } 201 201 + 202 202 + for _, msg := range unreadMessages { 203 203 + d.handleMessage(msg, convo) 204 204 + 205 205 + err = d.MarkMessageRead(msg.ID, convo.ID) 206 206 + if err != nil { 207 207 + slog.Error("marking message read", "error", err) 208 208 + continue 209 209 + } 210 210 + } 211 211 + } 212 212 + 213 213 + return nil 214 214 + } 215 215 + 216 216 + func (d *DmService) handleMessage(msg Message, convo Convo) { 217 217 + // TODO: add or remote user the list of "subsribed" users 218 218 + if strings.ToLower(msg.Text) == "subscribe" { 219 219 + userHandle := "" 220 220 + for _, member := range convo.Members { 221 221 + if member.Did == msg.Sender.Did { 222 222 + userHandle = member.Handle 223 223 + break 224 224 + } 225 225 + } 226 226 + 227 227 + if userHandle == "" { 228 228 + slog.Error("user handle for sent message not found", "sender did", msg.Sender.Did, "convo members", convo.Members) 229 229 + return 230 230 + } 231 231 + 232 232 + user := User{ 233 233 + DID: msg.Sender.Did, 234 234 + ConvoID: convo.ID, 235 235 + Handle: userHandle, 236 236 + CreatedAt: int(time.Now().UnixMilli()), 237 237 + } 238 238 + 239 239 + err := d.store.CreateUser(user) 240 240 + if err != nil { 241 241 + slog.Error("error creating user", "error", err, "user", user) 242 242 + return 243 243 + } 244 244 + } 245 245 + } 246 246 + 247 247 + func (d *DmService) GetUnreadMessages() (ListConvosResponse, error) { 248 248 + url := fmt.Sprintf("%s/xrpc/chat.bsky.convo.listConvos?readState=unread", d.pdsURL) 249 249 + request, err := http.NewRequest("GET", url, nil) 250 250 + if err != nil { 251 251 + return ListConvosResponse{}, fmt.Errorf("create new list convos http request: %w", err) 252 252 + } 253 253 + 254 254 + request.Header.Add("Content-Type", "application/json") 255 255 + request.Header.Add("Accept", "application/json") 256 256 + request.Header.Add("Atproto-Proxy", "did:web:api.bsky.chat#bsky_chat") 257 257 + request.Header.Add("Authorization", fmt.Sprintf("Bearer %s", d.auth.AccessJwt)) 258 258 + 259 259 + resp, err := d.httpClient.Do(request) 260 260 + if err != nil { 261 261 + return ListConvosResponse{}, fmt.Errorf("do http request to list convos: %w", err) 262 262 + } 263 263 + defer resp.Body.Close() 264 264 + 265 265 + if resp.StatusCode != http.StatusOK { 266 266 + var errorResp ErrorResponse 267 267 + err = decodeResp(resp.Body, &errorResp) 268 268 + if err != nil { 269 269 + return ListConvosResponse{}, err 270 270 + } 271 271 + 272 272 + return ListConvosResponse{}, fmt.Errorf("listing convos responded with code %d: %s", resp.StatusCode, errorResp.Error) 273 273 + } 274 274 + 275 275 + var listConvoResp ListConvosResponse 276 276 + err = decodeResp(resp.Body, &listConvoResp) 277 277 + if err != nil { 278 278 + return ListConvosResponse{}, err 279 279 + } 280 280 + 281 281 + return listConvoResp, nil 282 282 + } 283 283 + 284 284 + func (d *DmService) MarkMessageRead(messageID, convoID string) error { 285 285 + bodyReq := UpdateMessageReadRequest{ 286 286 + ConvoID: convoID, 287 287 + MessageID: messageID, 288 288 + } 289 289 + 290 290 + bodyB, err := json.Marshal(bodyReq) 291 291 + if err != nil { 292 292 + return fmt.Errorf("marshal update message request body: %w", err) 293 293 + } 294 294 + 295 295 + r := bytes.NewReader(bodyB) 296 296 + 297 297 + url := fmt.Sprintf("%s/xrpc/chat.bsky.convo.updateRead", d.pdsURL) 298 298 + request, err := http.NewRequest("POST", url, r) 299 299 + if err != nil { 300 300 + return fmt.Errorf("create new list convos http request: %w", err) 301 301 + } 302 302 + 303 303 + request.Header.Add("Content-Type", "application/json") 304 304 + request.Header.Add("Accept", "application/json") 305 305 + request.Header.Add("Atproto-Proxy", "did:web:api.bsky.chat#bsky_chat") 306 306 + request.Header.Add("Authorization", fmt.Sprintf("Bearer %s", d.auth.AccessJwt)) 307 307 + 308 308 + resp, err := d.httpClient.Do(request) 309 309 + if err != nil { 310 310 + return fmt.Errorf("do http request to update message read: %w", err) 311 311 + } 312 312 + defer resp.Body.Close() 313 313 + 314 314 + if resp.StatusCode == http.StatusOK { 315 315 + return nil 316 316 + } 317 317 + 318 318 + var errorResp ErrorResponse 319 319 + err = decodeResp(resp.Body, &errorResp) 320 320 + if err != nil { 321 321 + return err 322 322 + } 323 323 + 324 324 + return fmt.Errorf("listing convos responded with code %d: %s", resp.StatusCode, errorResp.Error) 325 325 + 326 326 + } 327 327 + 328 328 + func (d *DmService) Authenicate() (auth, error) { 329 329 + url := fmt.Sprintf("%s/com.atproto.server.createSession", baseBskyURL) 330 330 + 331 331 + requestData := map[string]interface{}{ 332 332 + "identifier": d.accessData.handle, 333 333 + "password": d.accessData.appPassword, 334 334 + } 335 335 + 336 336 + data, err := json.Marshal(requestData) 337 337 + if err != nil { 338 338 + return auth{}, errors.Wrap(err, "failed to marshal request") 339 339 + } 340 340 + 341 341 + r := bytes.NewReader(data) 342 342 + 343 343 + request, err := http.NewRequest("POST", url, r) 344 344 + if err != nil { 345 345 + return auth{}, errors.Wrap(err, "failed to create request") 346 346 + } 347 347 + 348 348 + request.Header.Add("Content-Type", "application/json") 349 349 + 350 350 + resp, err := d.httpClient.Do(request) 351 351 + if err != nil { 352 352 + return auth{}, errors.Wrap(err, "failed to make request") 353 353 + } 354 354 + defer resp.Body.Close() 355 355 + 356 356 + var loginResp auth 357 357 + err = decodeResp(resp.Body, &loginResp) 358 358 + if err != nil { 359 359 + return auth{}, err 360 360 + } 361 361 + 362 362 + return loginResp, nil 363 363 + } 364 364 + 365 365 + func (d *DmService) RefreshAuthenication(ctx context.Context) error { 366 366 + url := fmt.Sprintf("%s/com.atproto.server.refreshSession", baseBskyURL) 367 367 + 368 368 + request, err := http.NewRequest("POST", url, nil) 369 369 + if err != nil { 370 370 + return errors.Wrap(err, "failed to create request") 371 371 + } 372 372 + 373 373 + request.Header.Add("Content-Type", "application/json") 374 374 + request.Header.Add("Authorization", fmt.Sprintf("Bearer %s", d.auth.RefershJWT)) 375 375 + 376 376 + resp, err := d.httpClient.Do(request) 377 377 + if err != nil { 378 378 + return errors.Wrap(err, "failed to make request") 379 379 + } 380 380 + defer resp.Body.Close() 381 381 + 382 382 + var loginResp auth 383 383 + err = decodeResp(resp.Body, &loginResp) 384 384 + if err != nil { 385 385 + return err 386 386 + } 387 387 + 388 388 + d.auth = loginResp 389 389 + 390 390 + return nil 391 391 + } 392 392 + 393 393 + func (d *DmService) GetMessages(ctx context.Context, convoID string) (MessageResp, error) { 394 394 + url := fmt.Sprintf("%s/xrpc/chat.bsky.convo.getMessages?convoId=%s", d.pdsURL, convoID) 395 395 + request, err := http.NewRequest("GET", url, nil) 396 396 + if err != nil { 397 397 + return MessageResp{}, fmt.Errorf("create new get messages http request: %w", err) 398 398 + } 399 399 + 400 400 + request.Header.Add("Content-Type", "application/json") 401 401 + request.Header.Add("Accept", "application/json") 402 402 + request.Header.Add("Atproto-Proxy", "did:web:api.bsky.chat#bsky_chat") 403 403 + request.Header.Add("Authorization", fmt.Sprintf("Bearer %s", d.auth.AccessJwt)) 404 404 + 405 405 + resp, err := d.httpClient.Do(request) 406 406 + if err != nil { 407 407 + return MessageResp{}, fmt.Errorf("do http request to get messages: %w", err) 408 408 + } 409 409 + defer resp.Body.Close() 410 410 + 411 411 + if resp.StatusCode != http.StatusOK { 412 412 + var errorResp ErrorResponse 413 413 + err = decodeResp(resp.Body, &errorResp) 414 414 + if err != nil { 415 415 + return MessageResp{}, err 416 416 + } 417 417 + 418 418 + return MessageResp{}, fmt.Errorf("listing convos responded with code %d: %s", resp.StatusCode, errorResp.Error) 419 419 + } 420 420 + 421 421 + var messageResp MessageResp 422 422 + err = decodeResp(resp.Body, &messageResp) 423 423 + if err != nil { 424 424 + return MessageResp{}, err 425 425 + } 426 426 + 427 427 + return messageResp, nil 428 428 + } 429 429 + 430 430 + func decodeResp(body io.Reader, result any) error { 431 431 + resBody, err := io.ReadAll(body) 432 432 + if err != nil { 433 433 + return errors.Wrap(err, "failed to read response") 434 434 + } 435 435 + 436 436 + err = json.Unmarshal(resBody, result) 437 437 + if err != nil { 438 438 + return errors.Wrap(err, "failed to unmarshal response") 439 439 + } 440 440 + return nil 441 441 + }