Something went wrong. Try again.
A command-line interface for tangled, the decentralized Git collaboration platform built on the AT Protocol. eric-wien.tngl.io/tng-cli
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586package api
import ( "bytes" "context" "encoding/json" "fmt" "io" "net/http" "net/url" "strconv" "strings" "time"
"github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/auth/oauth" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax")
// defaultHTTPTimeout bounds knot/PDS reads made outside the DPoP client so a// slow or unresponsive server can't hang the CLI indefinitely.const defaultHTTPTimeout = 30 * time.Second
// maxBlobBytes caps blob downloads so a hostile PDS can't exhaust memory.const maxBlobBytes = 64 << 20 // 64 MiB
// Client wraps an authenticated session for making Tangled API calls.//// Tangled uses AT Protocol service auth: the CLI authenticates to the// user's PDS via OAuth/DPoP, then obtains short-lived service auth JWTs// for each target knot. Knots validate these JWTs via the user's DID// document signing key.type Client struct { session *oauth.ClientSession pdsClient *atclient.APIClient // authenticated against the PDS (DPoP) baseURL string // default appview URL (unused for XRPC, kept for web URLs) did syntax.DID dir identity.Directory // resolves handles/DIDs to PDS endpoints http *http.Client // for non-DPoP knot/PDS requests (bounded timeout)}
// NewClient creates a new API client from an OAuth session.func NewClient(session *oauth.ClientSession, baseURL string) *Client { did := session.Data.AccountDID if baseURL == "" { baseURL = "https://tangled.sh" } return &Client{ session: session, pdsClient: session.APIClient(), baseURL: baseURL, did: did, dir: identity.DefaultDirectory(), http: &http.Client{Timeout: defaultHTTPTimeout}, }}
// DID returns the authenticated user's DID.func (c *Client) DID() syntax.DID { return c.did}
// BaseURL returns the client's base URL.func (c *Client) BaseURL() string { return c.baseURL}
// PDSClient returns the underlying PDS API client for direct PDS calls.func (c *Client) PDSClient() *atclient.APIClient { return c.pdsClient}
// getServiceAuth obtains a service auth JWT from the PDS for the given// audience DID and XRPC method. The returned token is short-lived (~60s).func (c *Client) getServiceAuth(ctx context.Context, audienceDID, lxm string) (string, error) { nsid, err := syntax.ParseNSID("com.atproto.server.getServiceAuth") if err != nil { return "", err }
params := map[string]any{ "aud": audienceDID, "exp": time.Now().Unix() + 60, "lxm": lxm, }
var out struct { Token string `json:"token"` } if err := c.pdsClient.Get(ctx, nsid, params, &out); err != nil { return "", fmt.Errorf("failed to get service auth token: %w", err) } return out.Token, nil}
// ServiceAuthFor returns a Bearer token for making authenticated requests// to a knot. The knotHost is the knot's hostname (e.g. "knot.example.com").func (c *Client) ServiceAuthFor(ctx context.Context, knotHost, nsid string) (string, error) { audienceDID := "did:web:" + knotHost return c.getServiceAuth(ctx, audienceDID, nsid)}
// PostToKnot performs an authenticated XRPC procedure against a knot.// It obtains a service auth token from the PDS, then sends the request// with Bearer auth to the knot.func (c *Client) PostToKnot(ctx context.Context, knotURL, nsid string, input interface{}, out interface{}) error { knotHost := extractHost(knotURL) token, err := c.ServiceAuthFor(ctx, knotHost, nsid) if err != nil { return err }
u := fmt.Sprintf("%s/xrpc/%s", knotURL, nsid)
var body io.Reader if input != nil { data, err := json.Marshal(input) if err != nil { return fmt.Errorf("failed to marshal request body: %w", err) } body = bytes.NewReader(data) }
req, err := http.NewRequestWithContext(ctx, "POST", u, body) if err != nil { return err } if input != nil { req.Header.Set("Content-Type", "application/json") } req.Header.Set("Accept", "application/json") req.Header.Set("Authorization", "Bearer "+token)
return c.doPlainRequest(req, nsid, out)}
// GetFromKnot performs an unauthenticated XRPC query against a knot.// Most read endpoints on knots don't require auth.func (c *Client) GetFromKnot(ctx context.Context, knotURL, nsid string, params url.Values, out interface{}) error { u := fmt.Sprintf("%s/xrpc/%s", knotURL, nsid) if len(params) > 0 { u += "?" + params.Encode() }
req, err := http.NewRequestWithContext(ctx, "GET", u, nil) if err != nil { return err } req.Header.Set("Accept", "application/json")
return c.doPlainRequest(req, nsid, out)}
// PutRecord creates or updates an AT Protocol record on the user's PDS.func (c *Client) PutRecord(ctx context.Context, collection, rkey string, record interface{}) (string, error) { nsid, err := syntax.ParseNSID("com.atproto.repo.putRecord") if err != nil { return "", err }
body := map[string]interface{}{ "collection": collection, "repo": c.did.String(), "rkey": rkey, "record": record, }
var out struct { CID string `json:"cid"` URI string `json:"uri"` } if err := c.pdsClient.Post(ctx, nsid, body, &out); err != nil { return "", fmt.Errorf("failed to write record to PDS: %w", err) } return out.URI, nil}
// DeleteRecord deletes an AT Protocol record from the user's PDS.func (c *Client) DeleteRecord(ctx context.Context, collection, rkey string) error { nsid, err := syntax.ParseNSID("com.atproto.repo.deleteRecord") if err != nil { return err }
body := map[string]interface{}{ "collection": collection, "repo": c.did.String(), "rkey": rkey, }
if err := c.pdsClient.Post(ctx, nsid, body, nil); err != nil { return fmt.Errorf("failed to delete record from PDS: %w", err) } return nil}
// ListRecords lists AT Protocol records of a given collection from the PDS.func (c *Client) ListRecords(ctx context.Context, did, collection, cursor string, limit int) (*RecordList, error) { nsid, err := syntax.ParseNSID("com.atproto.repo.listRecords") if err != nil { return nil, err }
params := map[string]any{ "repo": did, "collection": collection, } if cursor != "" { params["cursor"] = cursor } if limit > 0 { params["limit"] = limit }
var out RecordList if err := c.pdsClient.Get(ctx, nsid, params, &out); err != nil { return nil, err } return &out, nil}
// ListPublicRecords lists AT Protocol records from another user's PDS.// It resolves the handle to a DID, then the DID to a PDS endpoint, and// queries that PDS directly. Used for cross-PDS lookups.func (c *Client) ListPublicRecords(ctx context.Context, repo, collection, cursor string, limit int) (*RecordList, error) { pdsURL, did, err := c.resolvePDS(ctx, repo) if err != nil { return nil, fmt.Errorf("failed to resolve PDS for %s: %w", repo, err) }
u := pdsURL + "/xrpc/com.atproto.repo.listRecords" params := url.Values{ "repo": {did}, "collection": {collection}, } if cursor != "" { params.Set("cursor", cursor) } if limit > 0 { params.Set("limit", strconv.Itoa(limit)) } u += "?" + params.Encode()
req, err := http.NewRequestWithContext(ctx, "GET", u, nil) if err != nil { return nil, err } req.Header.Set("Accept", "application/json")
var out RecordList if err := c.doPlainRequest(req, "com.atproto.repo.listRecords", &out); err != nil { return nil, err } return &out, nil}
// resolvePDS resolves a handle or DID to the PDS service endpoint URL and DID.// It uses indigo's identity directory, which handles handle resolution// (DNS/well-known) and all DID methods — did:plc via plc.directory and did:web// via the domain's /.well-known/did.json — rather than a hardcoded resolver.func (c *Client) resolvePDS(ctx context.Context, handleOrDID string) (string, string, error) { dir := c.dir if dir == nil { dir = identity.DefaultDirectory() }
id, err := syntax.ParseAtIdentifier(handleOrDID) if err != nil { return "", "", fmt.Errorf("invalid handle or DID %q: %w", handleOrDID, err) }
ident, err := dir.Lookup(ctx, id) if err != nil { return "", "", fmt.Errorf("failed to resolve identity for %s: %w", handleOrDID, err) }
pds := ident.PDSEndpoint() if pds == "" { return "", "", fmt.Errorf("no PDS endpoint in DID document for %s", handleOrDID) } return pds, ident.DID.String(), nil}
// RecordList is the response from com.atproto.repo.listRecords.type RecordList struct { Records []RecordEntry `json:"records"` Cursor string `json:"cursor,omitempty"`}
// RecordEntry is a single record from listRecords.type RecordEntry struct { URI string `json:"uri"` CID string `json:"cid"` Value json.RawMessage `json:"value"`}
// doPlainRequest executes an HTTP request without DPoP (used for knot requests// with Bearer service auth or unauthenticated reads).func (c *Client) doPlainRequest(req *http.Request, nsid string, out interface{}) error { httpClient := c.http if httpClient == nil { httpClient = http.DefaultClient } resp, err := httpClient.Do(req) if err != nil { return fmt.Errorf("request failed: %w", err) } defer resp.Body.Close()
if resp.StatusCode == http.StatusTooManyRequests { retryAfter := 60 * time.Second if ra := resp.Header.Get("Retry-After"); ra != "" { if secs, err := strconv.Atoi(ra); err == nil { retryAfter = time.Duration(secs) * time.Second } } apiErr := &APIError{StatusCode: resp.StatusCode, NSID: nsid} json.NewDecoder(resp.Body).Decode(apiErr) return &RateLimitError{RetryAfter: retryAfter, Inner: apiErr} }
if resp.StatusCode < 200 || resp.StatusCode >= 300 { apiErr := &APIError{StatusCode: resp.StatusCode, NSID: nsid} json.NewDecoder(resp.Body).Decode(apiErr) if apiErr.ErrorType == "" { apiErr.ErrorType = http.StatusText(resp.StatusCode) } return apiErr }
if out != nil { if err := json.NewDecoder(resp.Body).Decode(out); err != nil { return fmt.Errorf("failed to decode response: %w", err) } }
return nil}
// doRawRequest executes an HTTP request like doPlainRequest but returns the// raw response body (bounded by maxBlobBytes) instead of decoding JSON.func (c *Client) doRawRequest(req *http.Request, nsid string) ([]byte, error) { httpClient := c.http if httpClient == nil { httpClient = http.DefaultClient } resp, err := httpClient.Do(req) if err != nil { return nil, fmt.Errorf("request failed: %w", err) } defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 { apiErr := &APIError{StatusCode: resp.StatusCode, NSID: nsid} json.NewDecoder(resp.Body).Decode(apiErr) if apiErr.ErrorType == "" { apiErr.ErrorType = http.StatusText(resp.StatusCode) } return nil, apiErr }
data, err := io.ReadAll(io.LimitReader(resp.Body, maxBlobBytes+1)) if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) } if len(data) > maxBlobBytes { return nil, fmt.Errorf("response exceeds %d bytes", maxBlobBytes) } return data, nil}
// ResolvedRepo holds metadata about a Tangled repository resolved from AT Protocol records.type ResolvedRepo struct { URI string // at://did/sh.tangled.repo/rkey OwnerDID string Name string Knot string // knot hostname (e.g. "knot.example.com") KnotURL string // https://knot-host Rkey string}
// listRecordsAll walks every record of a collection for an owner, following// cursors. It tries the authenticated user's PDS first and, if that PDS doesn't// host the owner, falls back to the public cross-PDS path for the whole walk.// limit <= 0 fetches all records.func (c *Client) listRecordsAll(ctx context.Context, owner, collection string, limit int) ([]RecordEntry, error) { var all []RecordEntry cursor := "" usePublic := false
for { pageLimit := 100 if limit > 0 { remaining := limit - len(all) if remaining <= 0 { break } if remaining < pageLimit { pageLimit = remaining } }
var ( rl *RecordList err error ) if usePublic { rl, err = c.ListPublicRecords(ctx, owner, collection, cursor, pageLimit) } else { rl, err = c.ListRecords(ctx, owner, collection, cursor, pageLimit) if err != nil && cursor == "" { // The user's PDS may not host this owner — try the public API. usePublic = true rl, err = c.ListPublicRecords(ctx, owner, collection, cursor, pageLimit) } } if err != nil { return nil, err }
all = append(all, rl.Records...) // A repeated cursor means a misbehaving server; stop rather than loop forever. if rl.Cursor == "" || len(rl.Records) == 0 || rl.Cursor == cursor { break } cursor = rl.Cursor }
return all, nil}
// ResolveRepo looks up a repository by owner and name from AT Protocol records.// It first tries the authenticated user's PDS, then falls back to the public API// for cross-PDS lookups (e.g. resolving repos owned by users on a different PDS),// paging through all of the owner's repo records.func (c *Client) ResolveRepo(ctx context.Context, owner, name string) (*ResolvedRepo, error) { records, err := c.listRecordsAll(ctx, owner, RepoRecordNSID, 0) if err != nil { return nil, fmt.Errorf("failed to look up repos for %s: %w", owner, err) }
for _, rec := range records { var r struct { Name string `json:"name"` Knot string `json:"knot"` } if err := json.Unmarshal(rec.Value, &r); err != nil { continue } if r.Name == name { if r.Knot == "" { return nil, fmt.Errorf("repo %s/%s has no knot assigned", owner, name) } rkey := RkeyFromURI(rec.URI) ownerDID := DIDFromURI(rec.URI) return &ResolvedRepo{ URI: rec.URI, OwnerDID: ownerDID, Name: r.Name, Knot: r.Knot, KnotURL: "https://" + r.Knot, Rkey: rkey, }, nil } }
return nil, fmt.Errorf("repo %s/%s not found in AT Protocol records", owner, name)}
// GetRecord fetches a single AT Protocol record. It tries the authenticated// user's PDS first and falls back to the record owner's PDS for cross-PDS// reads — the user's PDS only serves repos it hosts.func (c *Client) GetRecord(ctx context.Context, did, collection, rkey string) (*RecordEntry, error) { nsid, err := syntax.ParseNSID("com.atproto.repo.getRecord") if err != nil { return nil, err }
params := map[string]any{ "repo": did, "collection": collection, "rkey": rkey, }
var out RecordEntry ownErr := c.pdsClient.Get(ctx, nsid, params, &out) if ownErr == nil { return &out, nil }
// The user's PDS may not host this DID — query the owner's PDS directly. pdsURL, resolvedDID, err := c.resolvePDS(ctx, did) if err != nil { return nil, ownErr } u := pdsURL + "/xrpc/com.atproto.repo.getRecord?" + url.Values{ "repo": {resolvedDID}, "collection": {collection}, "rkey": {rkey}, }.Encode()
req, err := http.NewRequestWithContext(ctx, "GET", u, nil) if err != nil { return nil, err } req.Header.Set("Accept", "application/json")
if err := c.doPlainRequest(req, "com.atproto.repo.getRecord", &out); err != nil { return nil, err } return &out, nil}
// UploadBlob uploads a blob to the user's PDS and returns the blob reference.// The returned json.RawMessage can be embedded directly in record fields.func (c *Client) UploadBlob(ctx context.Context, data io.Reader, contentType string) (json.RawMessage, error) { nsid, err := syntax.ParseNSID("com.atproto.repo.uploadBlob") if err != nil { return nil, err }
var out struct { Blob json.RawMessage `json:"blob"` } if err := c.pdsClient.LexDo(ctx, "POST", contentType, nsid.String(), nil, data, &out); err != nil { return nil, fmt.Errorf("failed to upload blob: %w", err) } return out.Blob, nil}
// RkeyFromURI extracts the rkey (last path segment) from an AT URI.func RkeyFromURI(uri string) string { parts := strings.Split(uri, "/") if len(parts) > 0 { return parts[len(parts)-1] } return ""}
// DIDFromURI extracts the DID from an AT URI (at://did/collection/rkey).func DIDFromURI(uri string) string { trimmed := strings.TrimPrefix(uri, "at://") if idx := strings.Index(trimmed, "/"); idx != -1 { return trimmed[:idx] } return trimmed}
// Paginate iterates through all pages of a paginated XRPC query.func Paginate[T any](ctx context.Context, fn func(cursor string) ([]T, string, error)) ([]T, error) { var all []T cursor := "" for { items, nextCursor, err := fn(cursor) if err != nil { return nil, err } all = append(all, items...) // A repeated cursor means a misbehaving server; stop rather than loop forever. if nextCursor == "" || nextCursor == cursor { break } cursor = nextCursor } return all, nil}
// extractHost extracts the hostname from a URL string.func extractHost(rawURL string) string { if !strings.Contains(rawURL, "://") { rawURL = "https://" + rawURL } u, err := url.Parse(rawURL) if err != nil { // Fallback: strip protocol prefix host := strings.TrimPrefix(rawURL, "https://") host = strings.TrimPrefix(host, "http://") if idx := strings.Index(host, "/"); idx != -1 { host = host[:idx] } return host } return u.Hostname()}