publish/pds_records.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
)

// BlobRef represents a blob reference for profile images
type BlobRef struct {
	Type     string `json:"$type"`
	Ref      Link   `json:"ref"`
	MimeType string `json:"mimeType"`
	Size     int64  `json:"size"`
}

type Link struct {
	Link string `json:"$link"`
}

// UploadBlob uploads an image to the PDS and returns a blob reference
func (p *Publisher) UploadBlob(session *PDSSession, data []byte, mimeType string) (*BlobRef, error) {
	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader(data))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", mimeType)
	req.Header.Set("Authorization", "Bearer "+session.AccessJwt)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("upload blob failed: %s - %s", resp.Status, string(respBody))
	}

	var result struct {
		Blob BlobRef `json:"blob"`
	}
	if err := json.Unmarshal(respBody, &result); err != nil {
		return nil, err
	}

	return &result.Blob, nil
}

// UpdateProfile updates the profile for an account
func (p *Publisher) UpdateProfile(session *PDSSession, displayName, description string, avatar *BlobRef) error {
	getReq, err := http.NewRequest("GET",
		p.pdsHost+"/xrpc/com.atproto.repo.getRecord?repo="+session.DID+"&collection=app.bsky.actor.profile&rkey=self",
		nil)
	if err != nil {
		return err
	}
	getReq.Header.Set("Authorization", "Bearer "+session.AccessJwt)

	getResp, err := p.httpClient.Do(getReq)

	var existingCID string
	profile := map[string]interface{}{
		"$type": "app.bsky.actor.profile",
	}

	if err == nil && getResp.StatusCode == http.StatusOK {
		defer getResp.Body.Close()
		var existing struct {
			CID   string                 `json:"cid"`
			Value map[string]interface{} `json:"value"`
		}
		if json.NewDecoder(getResp.Body).Decode(&existing) == nil {
			existingCID = existing.CID
			profile = existing.Value
		}
	} else if getResp != nil {
		getResp.Body.Close()
	}

	if displayName != "" {
		profile["displayName"] = displayName
	}
	if description != "" {
		profile["description"] = description
	}
	if avatar != nil {
		profile["avatar"] = avatar
	}

	payload := map[string]interface{}{
		"repo":       session.DID,
		"collection": "app.bsky.actor.profile",
		"rkey":       "self",
		"record":     profile,
	}
	if existingCID != "" {
		payload["swapRecord"] = existingCID
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.repo.putRecord", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+session.AccessJwt)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("update profile failed: %s - %s", resp.Status, string(respBody))
	}

	return nil
}

// DeleteAllPosts deletes all posts from an account
func (p *Publisher) DeleteAllPosts(session *PDSSession) (int, error) {
	deleted := 0
	cursor := ""

	for {
		listURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=app.bsky.feed.post&limit=100",
			p.pdsHost, session.DID)
		if cursor != "" {
			listURL += "&cursor=" + url.QueryEscape(cursor)
		}

		req, err := http.NewRequest("GET", listURL, nil)
		if err != nil {
			return deleted, err
		}
		req.Header.Set("Authorization", "Bearer "+session.AccessJwt)

		resp, err := p.httpClient.Do(req)
		if err != nil {
			return deleted, err
		}

		var result struct {
			Records []struct {
				URI string `json:"uri"`
			} `json:"records"`
			Cursor string `json:"cursor"`
		}

		respBody, _ := io.ReadAll(resp.Body)
		resp.Body.Close()

		if resp.StatusCode != http.StatusOK {
			return deleted, fmt.Errorf("list records failed: %s - %s", resp.Status, string(respBody))
		}

		if err := json.Unmarshal(respBody, &result); err != nil {
			return deleted, err
		}

		if len(result.Records) == 0 {
			break
		}

		for _, record := range result.Records {
			parts := strings.Split(record.URI, "/")
			if len(parts) < 2 {
				continue
			}
			rkey := parts[len(parts)-1]

			if err := p.DeleteRecord(session, "app.bsky.feed.post", rkey); err != nil {
				continue
			}
			deleted++
		}

		cursor = result.Cursor
		if cursor == "" {
			break
		}
	}

	return deleted, nil
}

// DeleteRecord deletes a single record from an account
func (p *Publisher) DeleteRecord(session *PDSSession, collection, rkey string) error {
	payload := map[string]interface{}{
		"repo":       session.DID,
		"collection": collection,
		"rkey":       rkey,
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.repo.deleteRecord", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+session.AccessJwt)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		respBody, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("delete record failed: %s - %s", resp.Status, string(respBody))
	}

	return nil
}

// DeleteAccount deletes an account using PDS admin API
func (p *Publisher) DeleteAccount(adminPassword, did string) error {
	payload := map[string]interface{}{
		"did": did,
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.admin.deleteAccount", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.SetBasicAuth("admin", adminPassword)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		respBody, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("delete account failed: %s - %s", resp.Status, string(respBody))
	}

	return nil
}

// TakedownAccount applies a takedown to an account
func (p *Publisher) TakedownAccount(adminPassword, did, reason string) error {
	payload := map[string]interface{}{
		"subject": map[string]interface{}{
			"$type": "com.atproto.admin.defs#repoRef",
			"did":   did,
		},
		"takedown": map[string]interface{}{
			"applied": true,
			"ref":     reason,
		},
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.admin.updateSubjectStatus", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.SetBasicAuth("admin", adminPassword)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		respBody, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("takedown account failed: %s - %s", resp.Status, string(respBody))
	}

	return nil
}

// RestoreAccount removes a takedown from an account
func (p *Publisher) RestoreAccount(adminPassword, did string) error {
	payload := map[string]interface{}{
		"subject": map[string]interface{}{
			"$type": "com.atproto.admin.defs#repoRef",
			"did":   did,
		},
		"takedown": map[string]interface{}{
			"applied": false,
		},
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.pdsHost+"/xrpc/com.atproto.admin.updateSubjectStatus", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.SetBasicAuth("admin", adminPassword)

	resp, err := p.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		respBody, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("restore account failed: %s - %s", resp.Status, string(respBody))
	}

	return nil
}