package pds import ( "context" "errors" "fmt" "net/http" "sort" "strconv" "strings" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/ipfs/go-cid" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/repo" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/httpjson" ) var errForbidden = errors.New("pds: forbidden") type recordView struct { URI string `json:"uri"` CID string `json:"cid"` Value map[string]any `json:"value"` } type listRecordsOutput struct { Records []recordView `json:"records"` Cursor string `json:"cursor,omitempty"` } type describeRepoOutput struct { DID string `json:"did"` Handle string `json:"handle"` HandleIsCorrect bool `json:"handleIsCorrect"` Collections []string `json:"collections"` DIDDoc any `json:"didDoc,omitempty"` } type commitView struct { CID string `json:"cid"` Rev string `json:"rev"` } type createRecordOutput struct { URI string `json:"uri"` CID string `json:"cid"` Commit *commitView `json:"commit,omitempty"` } type createWriteRequest struct { Repo string `json:"repo,omitempty"` Collection string `json:"collection"` Rkey string `json:"rkey,omitempty"` Record map[string]any `json:"record"` SwapCommit *string `json:"swapCommit,omitempty"` SwapRecord *string `json:"swapRecord,omitempty"` } type putWriteRequest struct { Repo string `json:"repo,omitempty"` Collection string `json:"collection"` Rkey string `json:"rkey"` Record map[string]any `json:"record"` SwapCommit *string `json:"swapCommit,omitempty"` SwapRecord *string `json:"swapRecord,omitempty"` } type deleteWriteRequest struct { Repo string `json:"repo,omitempty"` Collection string `json:"collection"` Rkey string `json:"rkey"` SwapCommit *string `json:"swapCommit,omitempty"` SwapRecord *string `json:"swapRecord,omitempty"` } type applyWrite struct { Type string `json:"$type"` Collection string `json:"collection"` Rkey string `json:"rkey,omitempty"` Value map[string]any `json:"value,omitempty"` SwapRecord *string `json:"swapRecord,omitempty"` } type applyWritesRequest struct { Repo string `json:"repo,omitempty"` Writes []applyWrite `json:"writes"` SwapCommit *string `json:"swapCommit,omitempty"` } type applyWritesOutput struct { Commit commitView `json:"commit"` Results []any `json:"results"` } type createWriteResult struct { Type string `json:"$type"` URI string `json:"uri"` CID string `json:"cid"` ValidationStatus string `json:"validationStatus,omitempty"` } type updateWriteResult struct { Type string `json:"$type"` URI string `json:"uri"` CID string `json:"cid"` ValidationStatus string `json:"validationStatus,omitempty"` } type deleteWriteResult struct { Type string `json:"$type"` } func (s *Server) handleGetRecord(w http.ResponseWriter, r *http.Request) { ctx := r.Context() query := r.URL.Query() rawRepo := query.Get("repo") if rawRepo == "" { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "repo is required") return } did, err := s.resolveRepo(ctx, rawRepo) if err != nil { s.repoError(w, err) return } collection, err := syntax.ParseNSID(query.Get("collection")) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } rkey, err := syntax.ParseRecordKey(query.Get("rkey")) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid rkey") return } rec, err := s.repos.GetRecord(ctx, did, collection, rkey) if err != nil { s.repoError(w, err) return } if rawCID := query.Get("cid"); rawCID != "" { want, err := cid.Parse(rawCID) if err != nil || !want.Equals(rec.CID) { writeXRPCError(w, http.StatusBadRequest, "RecordNotFound", "record cid does not match") return } } httpjson.Write(w, http.StatusOK, recordView{ URI: fmt.Sprintf("at://%s/%s/%s", did, collection, rkey), CID: rec.CID.String(), Value: rec.Value, }) } func (s *Server) handleListRecords(w http.ResponseWriter, r *http.Request) { ctx := r.Context() query := r.URL.Query() rawRepo := query.Get("repo") if rawRepo == "" { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "repo is required") return } did, err := s.resolveRepo(ctx, rawRepo) if err != nil { s.repoError(w, err) return } collection, err := syntax.ParseNSID(query.Get("collection")) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } limit := repo.DefaultList if rawLimit := query.Get("limit"); rawLimit != "" { n, err := strconv.Atoi(rawLimit) if err != nil || n < 1 || n > repo.MaxList { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "limit must be between 1 and 100") return } limit = n } var cursor syntax.RecordKey if rawCursor := query.Get("cursor"); rawCursor != "" { cursor, err = syntax.ParseRecordKey(rawCursor) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid cursor") return } } reverse := false if rawReverse := query.Get("reverse"); rawReverse != "" { reverse, err = strconv.ParseBool(rawReverse) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid reverse") return } } var records []repo.Record var next *syntax.RecordKey if reverse { records, next, err = s.listRecordsReverse(ctx, did, collection, cursor, limit) } else { records, next, err = s.repos.ListRecords(ctx, did, collection, cursor, limit) } if err != nil { s.repoError(w, err) return } out := listRecordsOutput{Records: make([]recordView, 0, len(records))} for _, rec := range records { out.Records = append(out.Records, recordView{ URI: fmt.Sprintf("at://%s/%s/%s", did, collection, rec.Rkey), CID: rec.CID.String(), Value: rec.Value, }) } if next != nil { out.Cursor = next.String() } httpjson.Write(w, http.StatusOK, out) } func (s *Server) listRecordsReverse(ctx context.Context, did syntax.DID, collection syntax.NSID, cursor syntax.RecordKey, limit int) ([]repo.Record, *syntax.RecordKey, error) { var pageCursor syntax.RecordKey all := make([]repo.Record, 0, limit) for { records, next, err := s.repos.ListRecords(ctx, did, collection, pageCursor, repo.MaxList) if err != nil { return nil, nil, err } stop := false for _, rec := range records { if cursor != "" && rec.Rkey.String() >= cursor.String() { stop = true continue } all = append(all, rec) } if stop || next == nil { break } pageCursor = *next if cursor != "" && pageCursor.String() >= cursor.String() { break } } if len(all) == 0 { return nil, nil, nil } start := max(len(all)-limit, 0) window := all[start:] out := make([]repo.Record, len(window)) for i := range window { out[len(window)-1-i] = window[i] } var next *syntax.RecordKey if start > 0 { rk := all[start].Rkey next = &rk } return out, next, nil } func (s *Server) handleDescribeRepo(w http.ResponseWriter, r *http.Request) { ctx := r.Context() rawRepo := r.URL.Query().Get("repo") if rawRepo == "" { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "repo is required") return } did, err := s.resolveRepo(ctx, rawRepo) if err != nil { s.repoError(w, err) return } handle, err := s.repos.OpenRepo(ctx, did, nil) if err != nil { s.repoError(w, err) return } collections := map[string]bool{} err = handle.Walk(ctx, "", func(path string, _ cid.Cid) error { i := strings.IndexByte(path, '/') if i <= 0 { return nil } nsid := path[:i] if _, err := syntax.ParseNSID(nsid); err == nil { collections[nsid] = true } return nil }) if err != nil { s.repoError(w, err) return } names := make([]string, 0, len(collections)) for name := range collections { names = append(names, name) } sort.Strings(names) handleName := s.accountHandle(ctx, did) httpjson.Write(w, http.StatusOK, describeRepoOutput{ DID: did.String(), Handle: handleName, HandleIsCorrect: s.handleIsCorrect(ctx, did, handleName), Collections: names, DIDDoc: s.didDocument(ctx, did), }) } func (s *Server) handleCreateRecord(w http.ResponseWriter, r *http.Request, info *authInfo) { ctx := r.Context() var req createWriteRequest if err := s.decodeJSON(w, r, &req); err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid JSON") return } collection, err := syntax.ParseNSID(req.Collection) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } if req.Record == nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "record is required") return } var rkey syntax.RecordKey if req.Rkey != "" { rkey, err = syntax.ParseRecordKey(req.Rkey) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid rkey") return } } swapCommit, err := parseCIDPtr(req.SwapCommit) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapCommit") return } did, handle, err := s.writeRepo(ctx, info, req.Repo) if err != nil { s.writeRepoError(w, err) return } ev, err := handle.ApplyWrites(ctx, []repo.WriteOp{{ Action: repo.ActionCreate, Collection: collection, Rkey: rkey, Record: recordType(req.Record, collection.String()), }}, swapCommit) if err != nil { s.repoError(w, err) return } op, ok := firstRecordOp(ev) if !ok || op.CID == nil { writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "write produced no record") return } httpjson.Write(w, http.StatusOK, createRecordOutput{ URI: uriForPath(did, op.Path), CID: op.CID.String(), Commit: commitFromEvent(ev), }) } func (s *Server) handlePutRecord(w http.ResponseWriter, r *http.Request, info *authInfo) { ctx := r.Context() var req putWriteRequest if err := s.decodeJSON(w, r, &req); err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid JSON") return } collection, err := syntax.ParseNSID(req.Collection) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } rkey, err := syntax.ParseRecordKey(req.Rkey) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid rkey") return } if req.Record == nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "record is required") return } swapCommit, err := parseCIDPtr(req.SwapCommit) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapCommit") return } swapRecord, err := parseCIDPtr(req.SwapRecord) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapRecord") return } did, handle, err := s.writeRepo(ctx, info, req.Repo) if err != nil { s.writeRepoError(w, err) return } ev, err := handle.ApplyWrites(ctx, []repo.WriteOp{{ Action: repo.WriteAction("put"), Collection: collection, Rkey: rkey, Record: recordType(req.Record, collection.String()), Prev: swapRecord, }}, swapCommit) if err != nil { s.repoError(w, err) return } op, ok := firstRecordOp(ev) if !ok || op.CID == nil { if rec, err := handle.GetRecord(ctx, collection, rkey); err == nil { httpjson.Write(w, http.StatusOK, createRecordOutput{ URI: fmt.Sprintf("at://%s/%s/%s", did, collection, rkey), CID: rec.CID.String(), Commit: commitFromEvent(ev), }) return } writeXRPCError(w, http.StatusInternalServerError, "InternalServerError", "write produced no record") return } httpjson.Write(w, http.StatusOK, createRecordOutput{ URI: uriForPath(did, op.Path), CID: op.CID.String(), Commit: commitFromEvent(ev), }) } func (s *Server) handleDeleteRecord(w http.ResponseWriter, r *http.Request, info *authInfo) { ctx := r.Context() var req deleteWriteRequest if err := s.decodeJSON(w, r, &req); err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid JSON") return } collection, err := syntax.ParseNSID(req.Collection) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } rkey, err := syntax.ParseRecordKey(req.Rkey) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid rkey") return } swapCommit, err := parseCIDPtr(req.SwapCommit) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapCommit") return } swapRecord, err := parseCIDPtr(req.SwapRecord) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapRecord") return } _, handle, err := s.writeRepo(ctx, info, req.Repo) if err != nil { s.writeRepoError(w, err) return } if _, err := handle.ApplyWrites(ctx, []repo.WriteOp{{ Action: repo.ActionDelete, Collection: collection, Rkey: rkey, Prev: swapRecord, }}, swapCommit); err != nil { s.repoError(w, err) return } httpjson.Write(w, http.StatusOK, map[string]any{}) } func (s *Server) handleApplyWrites(w http.ResponseWriter, r *http.Request, info *authInfo) { ctx := r.Context() var req applyWritesRequest if err := s.decodeJSON(w, r, &req); err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid JSON") return } if len(req.Writes) == 0 { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "writes is required") return } if len(req.Writes) > repo.MaxWrites { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "too many writes") return } swapCommit, err := parseCIDPtr(req.SwapCommit) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapCommit") return } writes := make([]repo.WriteOp, 0, len(req.Writes)) inputActions := make([]repo.WriteAction, 0, len(req.Writes)) inputRkeys := make([]syntax.RecordKey, 0, len(req.Writes)) for _, write := range req.Writes { kind := write.Type if i := strings.LastIndexByte(kind, '#'); i >= 0 { kind = kind[i+1:] } collection, err := syntax.ParseNSID(write.Collection) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid collection") return } var rkey syntax.RecordKey if write.Rkey != "" { rkey, err = syntax.ParseRecordKey(write.Rkey) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid rkey") return } } var action repo.WriteAction switch kind { case "create": action = repo.ActionCreate if write.Value == nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "create write requires value") return } case "update": action = repo.ActionUpdate if rkey == "" { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "update write requires rkey") return } if write.Value == nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "update write requires value") return } case "delete": action = repo.ActionDelete if rkey == "" { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "delete write requires rkey") return } if write.Value != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "delete write must not include value") return } default: writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "unknown write type") return } var prev *cid.Cid if write.SwapRecord != nil { prev, err = parseCIDPtr(write.SwapRecord) if err != nil { writeXRPCError(w, http.StatusBadRequest, "InvalidRequest", "invalid swapRecord") return } } op := repo.WriteOp{ Action: action, Collection: collection, Rkey: rkey, Prev: prev, } if action != repo.ActionDelete { op.Record = recordType(write.Value, collection.String()) } writes = append(writes, op) inputActions = append(inputActions, action) inputRkeys = append(inputRkeys, rkey) } did, handle, err := s.writeRepo(ctx, info, req.Repo) if err != nil { s.writeRepoError(w, err) return } ev, err := handle.ApplyWrites(ctx, writes, swapCommit) if err != nil { s.repoError(w, err) return } results, err := s.applyWriteResults(ctx, did, handle, writes, inputActions, inputRkeys, ev) if err != nil { s.repoError(w, err) return } out := applyWritesOutput{Results: results} if commit := commitFromEvent(ev); commit != nil { out.Commit = *commit } httpjson.Write(w, http.StatusOK, out) } func (s *Server) applyWriteResults(ctx context.Context, did syntax.DID, handle repo.Repo, writes []repo.WriteOp, actions []repo.WriteAction, rkeys []syntax.RecordKey, ev *repo.CommitEvent) ([]any, error) { used := make([]bool, len(ev.Ops)) results := make([]any, len(writes)) for i, action := range actions { if action == repo.ActionDelete { results[i] = deleteWriteResult{Type: "com.atproto.repo.applyWrites#deleteResult"} continue } path := writes[i].Collection.String() + "/" + rkeys[i].String() found := -1 for j := range ev.Ops { if used[j] || ev.Ops[j].Action == repo.ActionDelete { continue } if rkeys[i] != "" { if ev.Ops[j].Path == path { found = j break } } else if ev.Ops[j].Action == repo.ActionCreate && strings.HasPrefix(ev.Ops[j].Path, writes[i].Collection.String()+"/") { found = j break } } resultPath := path var resultCID *cid.Cid resultAction := action if found >= 0 { used[found] = true resultPath = ev.Ops[found].Path resultCID = ev.Ops[found].CID resultAction = ev.Ops[found].Action } else if rkeys[i] != "" { rec, err := handle.GetRecord(ctx, writes[i].Collection, rkeys[i]) if err != nil { return nil, err } resultCID = &rec.CID } else { return nil, fmt.Errorf("%w: create produced no record op", repo.ErrInvalidRequest) } cidString := "" if resultCID != nil { cidString = resultCID.String() } uri := uriForPath(did, resultPath) if resultAction == repo.ActionCreate { results[i] = createWriteResult{ Type: "com.atproto.repo.applyWrites#createResult", URI: uri, CID: cidString, ValidationStatus: "valid", } } else { results[i] = updateWriteResult{ Type: "com.atproto.repo.applyWrites#updateResult", URI: uri, CID: cidString, ValidationStatus: "valid", } } } return results, nil } func (s *Server) writeRepo(ctx context.Context, info *authInfo, raw string) (syntax.DID, repo.Repo, error) { if raw == "" { raw = info.DID.String() } did, err := s.resolveRepo(ctx, raw) if err != nil { return "", nil, err } if did != info.DID { return "", nil, errForbidden } handle, err := s.openRepo(ctx, did) if err != nil { return "", nil, err } return did, handle, nil } func (s *Server) writeRepoError(w http.ResponseWriter, err error) { if errors.Is(err, errForbidden) { writeXRPCError(w, http.StatusForbidden, "Forbidden", "cannot write to another account") return } s.repoError(w, err) } func firstRecordOp(ev *repo.CommitEvent) (*repo.CommitOp, bool) { if ev == nil { return nil, false } for i := range ev.Ops { op := &ev.Ops[i] if op.Action != repo.ActionDelete { return op, true } } return nil, false } func commitFromEvent(ev *repo.CommitEvent) *commitView { if ev == nil { return nil } return &commitView{CID: ev.Commit.String(), Rev: ev.Rev.String()} } func uriForPath(did syntax.DID, path string) string { return "at://" + did.String() + "/" + path } func parseCIDPtr(raw *string) (*cid.Cid, error) { if raw == nil || *raw == "" { return nil, nil } c, err := cid.Parse(*raw) if err != nil { return nil, err } return &c, nil }