mirrors/at-container-registry
1
0
Fork 0
mirror of https://tangled.org/evan.jarrett.net/at-container-registry synced 2026-09-19 16:54:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

start researching quotas based on layer size per DID

Browse Source
This commit is contained in:
Evan Jarrett
2026-01-04 15:53:44 -06:00
parent f74bc3018a
commit 53e196a261
15 changed files with 886 additions and 1344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/QUOTAS.md
+364 -1113
View File
File diff suppressed because it is too large Load Diff
lexicons/io/atcr/hold/layer.json
+4 -9
View File
@@ -8,7 +8,7 @@
"description": "Represents metadata about a container layer stored in the hold. Stored in the hold's embedded PDS for tracking and analytics.",
"record": {
"type": "object",
"required": ["digest", "size", "mediaType", "repository", "userDid", "userHandle", "createdAt"],
"required": ["digest", "size", "mediaType", "manifest", "userDid", "createdAt"],
"properties": {
"digest": {
"type": "string",
@@ -24,21 +24,16 @@
"description": "Media type (e.g., application/vnd.oci.image.layer.v1.tar+gzip)",
"maxLength": 128
},
"repository": {
"manifest": {
"type": "string",
"description": "Repository this layer belongs to",
"maxLength": 255
"format": "at-uri",
"description": "AT-URI of the manifest that included this layer (e.g., at://did:plc:xyz/io.atcr.manifest/abc123)"
},
"userDid": {
"type": "string",
"format": "did",
"description": "DID of user who uploaded this layer"
},
"userHandle": {
"type": "string",
"format": "handle",
"description": "Handle of user (for display purposes)"
},
"createdAt": {
"type": "string",
"format": "datetime",
pkg/appview/handlers/storage.go
+129
View File
@@ -0,0 +1,129 @@
package handlers
import (
"encoding/json"
"fmt"
"html/template"
"log/slog"
"net/http"
"atcr.io/pkg/appview/middleware"
"atcr.io/pkg/appview/storage"
"atcr.io/pkg/atproto"
"atcr.io/pkg/auth/oauth"
)
// StorageHandler handles the storage quota API endpoint
// Returns an HTML partial for HTMX to swap into the settings page
type StorageHandler struct {
Templates *template.Template
Refresher *oauth.Refresher
}
// QuotaStats mirrors the hold service response
type QuotaStats struct {
UserDID string `json:"userDid"`
UniqueBlobs int `json:"uniqueBlobs"`
TotalSize int64 `json:"totalSize"`
}
func (h *StorageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
user := middleware.GetUser(r)
if user == nil {
http.Error(w, "Unauthorized", http.StatusUnauthorized)
return
}
// Create ATProto client with session provider
client := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher)
// Get user's sailor profile to find their default hold
profile, err := storage.GetProfile(r.Context(), client)
if err != nil {
slog.Warn("Failed to get profile for storage quota", "did", user.DID, "error", err)
h.renderError(w, "Failed to load profile")
return
}
if profile == nil || profile.DefaultHold == "" {
// No default hold configured - can't check quota
h.renderNoHold(w)
return
}
// Resolve hold URL from DID
holdURL := atproto.ResolveHoldURL(profile.DefaultHold)
if holdURL == "" {
slog.Warn("Failed to resolve hold URL", "did", user.DID, "holdDid", profile.DefaultHold)
h.renderError(w, "Failed to resolve hold service")
return
}
// Call the hold's quota endpoint
quotaURL := fmt.Sprintf("%s%s?userDid=%s", holdURL, atproto.HoldGetQuota, user.DID)
resp, err := http.Get(quotaURL)
if err != nil {
slog.Warn("Failed to fetch quota from hold", "did", user.DID, "holdURL", holdURL, "error", err)
h.renderError(w, "Failed to connect to hold service")
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
slog.Warn("Hold returned error for quota", "did", user.DID, "status", resp.StatusCode)
h.renderError(w, "Hold service returned an error")
return
}
var stats QuotaStats
if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil {
slog.Warn("Failed to decode quota response", "did", user.DID, "error", err)
h.renderError(w, "Failed to parse quota data")
return
}
// Render the stats partial
h.renderStats(w, stats)
}
func (h *StorageHandler) renderStats(w http.ResponseWriter, stats QuotaStats) {
data := struct {
UniqueBlobs int
TotalSize int64
HumanSize string
}{
UniqueBlobs: stats.UniqueBlobs,
TotalSize: stats.TotalSize,
HumanSize: humanizeBytes(stats.TotalSize),
}
w.Header().Set("Content-Type", "text/html")
if err := h.Templates.ExecuteTemplate(w, "storage_stats", data); err != nil {
slog.Error("Failed to render storage stats template", "error", err)
http.Error(w, "Failed to render template", http.StatusInternalServerError)
}
}
func (h *StorageHandler) renderError(w http.ResponseWriter, message string) {
w.Header().Set("Content-Type", "text/html")
fmt.Fprintf(w, `<div class="storage-error"><i data-lucide="alert-circle"></i> %s</div>`, message)
}
func (h *StorageHandler) renderNoHold(w http.ResponseWriter) {
w.Header().Set("Content-Type", "text/html")
fmt.Fprint(w, `<div class="storage-info"><i data-lucide="info"></i> No hold configured. Set a default hold above to see storage usage.</div>`)
}
// humanizeBytes converts bytes to human-readable format
func humanizeBytes(bytes int64) string {
const unit = 1024
if bytes < unit {
return fmt.Sprintf("%d B", bytes)
}
div, exp := int64(unit), 0
for n := bytes / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.1f %cB", float64(bytes)/float64(div), "KMGTPE"[exp])
}
pkg/appview/routes/routes.go
+5
View File
@@ -174,6 +174,11 @@ func RegisterUIRoutes(router chi.Router, deps UIDependencies) {
RegistryURL: registryURL,
}).ServeHTTP)
r.Get("/api/storage", (&uihandlers.StorageHandler{
Templates: deps.Templates,
Refresher: deps.Refresher,
}).ServeHTTP)
r.Post("/api/profile/default-hold", (&uihandlers.UpdateDefaultHoldHandler{
Refresher: deps.Refresher,
}).ServeHTTP)
pkg/appview/storage/manifest_store.go
+5 -4
View File
@@ -325,11 +325,12 @@ func (s *ManifestStore) notifyHoldAboutManifest(ctx context.Context, manifestRec
serviceToken := s.ctx.ServiceToken
// Build notification request
// Note: userHandle is resolved from userDid on the hold side (cached, 24-hour TTL)
notifyReq := map[string]any{
"repository": s.ctx.Repository,
"userDid": s.ctx.DID,
"userHandle": s.ctx.Handle,
"operation": operation,
"repository": s.ctx.Repository,
"userDid": s.ctx.DID,
"manifestDigest": manifestDigest,
"operation": operation,
}
// For push operations, include full manifest data
pkg/appview/templates/pages/settings.html
+59
View File
@@ -29,6 +29,15 @@
</div>
</section>
<!-- Storage Usage Section -->
<section class="settings-section storage-section">
<h2>Storage Usage</h2>
<p>Estimated storage usage on your default hold.</p>
<div id="storage-stats" hx-get="/api/storage" hx-trigger="load" hx-swap="innerHTML">
<p><i data-lucide="loader-2" class="spin"></i> Loading...</p>
</div>
</section>
<!-- Default Hold Section -->
<section class="settings-section">
<h2>Default Hold</h2>
@@ -200,6 +209,56 @@
</script>
<style>
/* Storage Section Styles */
.storage-section .storage-stats {
background: var(--code-bg);
padding: 1rem;
border-radius: 4px;
margin-top: 0.5rem;
}
.storage-section .stat-row {
display: flex;
justify-content: space-between;
padding: 0.5rem 0;
border-bottom: 1px solid var(--border);
}
.storage-section .stat-row:last-child {
border-bottom: none;
}
.storage-section .stat-label {
color: var(--fg-muted);
}
.storage-section .stat-value {
font-weight: bold;
font-family: monospace;
}
.storage-section .storage-error,
.storage-section .storage-info {
padding: 1rem;
border-radius: 4px;
margin-top: 0.5rem;
display: flex;
align-items: center;
gap: 0.5rem;
}
.storage-section .storage-error {
background: var(--error-bg, #fef2f2);
color: var(--error, #dc2626);
border: 1px solid var(--error, #dc2626);
}
.storage-section .storage-info {
background: var(--info-bg, #eff6ff);
color: var(--info, #2563eb);
border: 1px solid var(--info, #2563eb);
}
.spin {
animation: spin 1s linear infinite;
}
@keyframes spin {
from { transform: rotate(0deg); }
to { transform: rotate(360deg); }
}
/* Devices Section Styles */
.devices-section .setup-instructions {
margin: 1rem 0;
pkg/appview/templates/partials/storage_stats.html
+12
View File
@@ -0,0 +1,12 @@
{{ define "storage_stats" }}
<div class="storage-stats">
<div class="stat-row">
<span class="stat-label">Unique Blobs:</span>
<span class="stat-value">{{ .UniqueBlobs }}</span>
</div>
<div class="stat-row">
<span class="stat-label">Total Storage:</span>
<span class="stat-value">{{ .HumanSize }}</span>
</div>
</div>
{{ end }}
pkg/atproto/cbor_gen.go
+36 -70
View File
@@ -654,7 +654,7 @@ func (t *LayerRecord) MarshalCBOR(w io.Writer) error {
cw := cbg.NewCborWriter(w)
if _, err := cw.Write([]byte{168}); err != nil {
if _, err := cw.Write([]byte{167}); err != nil {
return err
}
@@ -749,6 +749,29 @@ func (t *LayerRecord) MarshalCBOR(w io.Writer) error {
return err
}
// t.Manifest (string) (string)
if len("manifest") > 8192 {
return xerrors.Errorf("Value in field \"manifest\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("manifest"))); err != nil {
return err
}
if _, err := cw.WriteString(string("manifest")); err != nil {
return err
}
if len(t.Manifest) > 8192 {
return xerrors.Errorf("Value in field t.Manifest was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Manifest))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Manifest)); err != nil {
return err
}
// t.CreatedAt (string) (string)
if len("createdAt") > 8192 {
return xerrors.Errorf("Value in field \"createdAt\" was too long")
@@ -794,52 +817,6 @@ func (t *LayerRecord) MarshalCBOR(w io.Writer) error {
if _, err := cw.WriteString(string(t.MediaType)); err != nil {
return err
}
// t.Repository (string) (string)
if len("repository") > 8192 {
return xerrors.Errorf("Value in field \"repository\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("repository"))); err != nil {
return err
}
if _, err := cw.WriteString(string("repository")); err != nil {
return err
}
if len(t.Repository) > 8192 {
return xerrors.Errorf("Value in field t.Repository was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Repository))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.Repository)); err != nil {
return err
}
// t.UserHandle (string) (string)
if len("userHandle") > 8192 {
return xerrors.Errorf("Value in field \"userHandle\" was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("userHandle"))); err != nil {
return err
}
if _, err := cw.WriteString(string("userHandle")); err != nil {
return err
}
if len(t.UserHandle) > 8192 {
return xerrors.Errorf("Value in field t.UserHandle was too long")
}
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.UserHandle))); err != nil {
return err
}
if _, err := cw.WriteString(string(t.UserHandle)); err != nil {
return err
}
return nil
}
@@ -868,7 +845,7 @@ func (t *LayerRecord) UnmarshalCBOR(r io.Reader) (err error) {
n := extra
nameBuf := make([]byte, 10)
nameBuf := make([]byte, 9)
for i := uint64(0); i < n; i++ {
nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192)
if err != nil {
@@ -943,6 +920,17 @@ func (t *LayerRecord) UnmarshalCBOR(r io.Reader) (err error) {
t.UserDID = string(sval)
}
// t.Manifest (string) (string)
case "manifest":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Manifest = string(sval)
}
// t.CreatedAt (string) (string)
case "createdAt":
@@ -965,28 +953,6 @@ func (t *LayerRecord) UnmarshalCBOR(r io.Reader) (err error) {
t.MediaType = string(sval)
}
// t.Repository (string) (string)
case "repository":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.Repository = string(sval)
}
// t.UserHandle (string) (string)
case "userHandle":
{
sval, err := cbg.ReadStringWithMax(cr, 8192)
if err != nil {
return err
}
t.UserHandle = string(sval)
}
default:
// Field doesn't exist on this type, so ignore it
pkg/atproto/endpoints.go
+6
View File
@@ -51,6 +51,12 @@ const (
// Request: {"ownerDid": "...", "repository": "...", "pullCount": 10, "pushCount": 5, "lastPull": "...", "lastPush": "..."}
// Response: {"success": true}
HoldSetStats = "/xrpc/io.atcr.hold.setStats"
// HoldGetQuota returns storage quota information for a user.
// Method: GET
// Query: userDid={did}
// Response: {"userDid": "...", "uniqueBlobs": 10, "totalSize": 1073741824}
HoldGetQuota = "/xrpc/io.atcr.hold.getQuota"
)
// Hold service crew management endpoints (io.atcr.hold.*)
pkg/atproto/lexicon.go
+16 -17
View File
@@ -602,27 +602,26 @@ type CrewRecord struct {
// Stored in the hold's embedded PDS for tracking and analytics
// Uses CBOR encoding for efficient storage in hold's carstore
type LayerRecord struct {
Type string `json:"$type" cborgen:"$type"`
Digest string `json:"digest" cborgen:"digest"` // Layer digest (e.g., "sha256:abc123...")
Size int64 `json:"size" cborgen:"size"` // Size in bytes
MediaType string `json:"mediaType" cborgen:"mediaType"` // Media type (e.g., "application/vnd.oci.image.layer.v1.tar+gzip")
Repository string `json:"repository" cborgen:"repository"` // Repository this layer belongs to
UserDID string `json:"userDid" cborgen:"userDid"` // DID of user who uploaded this layer
UserHandle string `json:"userHandle" cborgen:"userHandle"` // Handle of user (for display purposes)
CreatedAt string `json:"createdAt" cborgen:"createdAt"` // RFC3339 timestamp
Type string `json:"$type" cborgen:"$type"`
Digest string `json:"digest" cborgen:"digest"` // Layer digest (e.g., "sha256:abc123...")
Size int64 `json:"size" cborgen:"size"` // Size in bytes
MediaType string `json:"mediaType" cborgen:"mediaType"` // Media type (e.g., "application/vnd.oci.image.layer.v1.tar+gzip")
Manifest string `json:"manifest" cborgen:"manifest"` // AT-URI of manifest that included this layer
UserDID string `json:"userDid" cborgen:"userDid"` // DID of user who uploaded this layer
CreatedAt string `json:"createdAt" cborgen:"createdAt"` // RFC3339 timestamp
}
// NewLayerRecord creates a new layer record
func NewLayerRecord(digest string, size int64, mediaType, repository, userDID, userHandle string) *LayerRecord {
// manifestURI: AT-URI of the manifest (e.g., "at://did:plc:xyz/io.atcr.manifest/abc123")
func NewLayerRecord(digest string, size int64, mediaType, userDID, manifestURI string) *LayerRecord {
return &LayerRecord{
Type: LayerCollection,
Digest: digest,
Size: size,
MediaType: mediaType,
Repository: repository,
UserDID: userDID,
UserHandle: userHandle,
CreatedAt: time.Now().Format(time.RFC3339),
Type: LayerCollection,
Digest: digest,
Size: size,
MediaType: mediaType,
Manifest: manifestURI,
UserDID: userDID,
CreatedAt: time.Now().Format(time.RFC3339),
}
}
pkg/atproto/lexicon_test.go
+36 -49
View File
@@ -1089,55 +1089,50 @@ func TestRepositoryTagRoundTrip(t *testing.T) {
func TestNewLayerRecord(t *testing.T) {
tests := []struct {
name string
digest string
size int64
mediaType string
repository string
userDID string
userHandle string
name string
digest string
size int64
mediaType string
userDID string
manifestURI string
}{
{
name: "standard layer",
digest: "sha256:abc123",
size: 1024,
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
repository: "myapp",
userDID: "did:plc:user123",
userHandle: "alice.bsky.social",
name: "standard layer",
digest: "sha256:abc123",
size: 1024,
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
userDID: "did:plc:user123",
manifestURI: "at://did:plc:user123/io.atcr.manifest/abc123",
},
{
name: "large layer",
digest: "sha256:def456",
size: 1073741824, // 1GB
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
repository: "largeapp",
userDID: "did:plc:user456",
userHandle: "bob.example.com",
name: "large layer",
digest: "sha256:def456",
size: 1073741824, // 1GB
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
userDID: "did:plc:user456",
manifestURI: "at://did:plc:user456/io.atcr.manifest/def456",
},
{
name: "empty values",
digest: "",
size: 0,
mediaType: "",
repository: "",
userDID: "",
userHandle: "",
name: "empty values",
digest: "",
size: 0,
mediaType: "",
userDID: "",
manifestURI: "",
},
{
name: "config layer",
digest: "sha256:config123",
size: 512,
mediaType: "application/vnd.oci.image.config.v1+json",
repository: "app/subapp",
userDID: "did:web:example.com",
userHandle: "charlie.tangled.io",
name: "config layer",
digest: "sha256:config123",
size: 512,
mediaType: "application/vnd.oci.image.config.v1+json",
userDID: "did:web:example.com",
manifestURI: "at://did:web:example.com/io.atcr.manifest/config123",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
record := NewLayerRecord(tt.digest, tt.size, tt.mediaType, tt.repository, tt.userDID, tt.userHandle)
record := NewLayerRecord(tt.digest, tt.size, tt.mediaType, tt.userDID, tt.manifestURI)
// Verify all fields
if record == nil {
@@ -1160,18 +1155,14 @@ func TestNewLayerRecord(t *testing.T) {
t.Errorf("MediaType = %q, want %q", record.MediaType, tt.mediaType)
}
if record.Repository != tt.repository {
t.Errorf("Repository = %q, want %q", record.Repository, tt.repository)
if record.Manifest != tt.manifestURI {
t.Errorf("Manifest = %q, want %q", record.Manifest, tt.manifestURI)
}
if record.UserDID != tt.userDID {
t.Errorf("UserDID = %q, want %q", record.UserDID, tt.userDID)
}
if record.UserHandle != tt.userHandle {
t.Errorf("UserHandle = %q, want %q", record.UserHandle, tt.userHandle)
}
// Verify CreatedAt is set and is a valid RFC3339 timestamp
if record.CreatedAt == "" {
t.Error("CreatedAt is empty")
@@ -1192,9 +1183,8 @@ func TestNewLayerRecordJSON(t *testing.T) {
"sha256:abc123",
1024,
"application/vnd.oci.image.layer.v1.tar+gzip",
"myapp",
"did:plc:user123",
"alice.bsky.social",
"at://did:plc:user123/io.atcr.manifest/abc123",
)
// Marshal to JSON
@@ -1222,15 +1212,12 @@ func TestNewLayerRecordJSON(t *testing.T) {
if decoded.MediaType != record.MediaType {
t.Errorf("MediaType = %q, want %q", decoded.MediaType, record.MediaType)
}
if decoded.Repository != record.Repository {
t.Errorf("Repository = %q, want %q", decoded.Repository, record.Repository)
if decoded.Manifest != record.Manifest {
t.Errorf("Manifest = %q, want %q", decoded.Manifest, record.Manifest)
}
if decoded.UserDID != record.UserDID {
t.Errorf("UserDID = %q, want %q", decoded.UserDID, record.UserDID)
}
if decoded.UserHandle != record.UserHandle {
t.Errorf("UserHandle = %q, want %q", decoded.UserHandle, record.UserHandle)
}
if decoded.CreatedAt != record.CreatedAt {
t.Errorf("CreatedAt = %q, want %q", decoded.CreatedAt, record.CreatedAt)
}
pkg/hold/oci/xrpc.go
+18 -9
View File
@@ -217,12 +217,12 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
// Parse request
var req struct {
Repository string `json:"repository"`
Tag string `json:"tag"`
UserDID string `json:"userDid"`
UserHandle string `json:"userHandle"`
Operation string `json:"operation"` // "push" or "pull", defaults to "push" for backward compatibility
Manifest struct {
Repository string `json:"repository"`
Tag string `json:"tag"`
UserDID string `json:"userDid"`
ManifestDigest string `json:"manifestDigest"` // For building layer record AT-URIs
Operation string `json:"operation"` // "push" or "pull", defaults to "push" for backward compatibility
Manifest struct {
MediaType string `json:"mediaType"`
Config struct {
Digest string `json:"digest"`
@@ -287,15 +287,17 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
postsEnabled = h.enableBlueskyPosts
}
// Build manifest AT-URI for layer records
manifestURI := atproto.BuildManifestURI(req.UserDID, req.ManifestDigest)
// Create layer records for each blob
for _, layer := range req.Manifest.Layers {
record := atproto.NewLayerRecord(
layer.Digest,
layer.Size,
layer.MediaType,
req.Repository,
req.UserDID,
req.UserHandle,
manifestURI,
)
_, _, err := h.pds.CreateLayerRecord(ctx, record)
@@ -329,6 +331,13 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
// Create Bluesky post if enabled
if postsEnabled {
// Resolve handle from DID (cached, 24-hour TTL)
_, userHandle, _, resolveErr := atproto.ResolveIdentity(ctx, req.UserDID)
if resolveErr != nil {
slog.Warn("Failed to resolve handle for user", "did", req.UserDID, "error", resolveErr)
userHandle = req.UserDID // Fallback to DID if resolution fails
}
// Extract manifest digest from first layer (or use config digest as fallback)
manifestDigest := req.Manifest.Config.Digest
if len(req.Manifest.Layers) > 0 {
@@ -340,7 +349,7 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques
h.driver,
req.Repository,
req.Tag,
req.UserHandle,
userHandle,
req.UserDID,
manifestDigest,
totalSize,
pkg/hold/pds/layer.go
+103
View File
@@ -5,6 +5,8 @@ import (
"fmt"
"atcr.io/pkg/atproto"
lexutil "github.com/bluesky-social/indigo/lex/util"
"github.com/bluesky-social/indigo/repo"
)
// CreateLayerRecord creates a new layer record in the hold's PDS
@@ -57,3 +59,104 @@ func (p *HoldPDS) ListLayerRecords(ctx context.Context, limit int, cursor string
// not for runtime queries
return nil, "", fmt.Errorf("ListLayerRecords not yet implemented")
}
// QuotaStats represents storage quota information for a user
type QuotaStats struct {
UserDID string `json:"userDid"`
UniqueBlobs int `json:"uniqueBlobs"`
TotalSize int64 `json:"totalSize"`
}
// GetQuotaForUser calculates storage quota for a specific user
// It iterates through all layer records, filters by userDid, deduplicates by digest,
// and sums the sizes of unique blobs.
func (p *HoldPDS) GetQuotaForUser(ctx context.Context, userDID string) (*QuotaStats, error) {
if p.recordsIndex == nil {
return nil, fmt.Errorf("records index not available")
}
// Get session for reading record data
session, err := p.carstore.ReadOnlySession(p.uid)
if err != nil {
return nil, fmt.Errorf("failed to create session: %w", err)
}
head, err := p.carstore.GetUserRepoHead(ctx, p.uid)
if err != nil {
return nil, fmt.Errorf("failed to get repo head: %w", err)
}
if !head.Defined() {
// Empty repo - return zero stats
return &QuotaStats{UserDID: userDID}, nil
}
repoHandle, err := repo.OpenRepo(ctx, session, head)
if err != nil {
return nil, fmt.Errorf("failed to open repo: %w", err)
}
// Track unique digests and their sizes
digestSizes := make(map[string]int64)
// Iterate all layer records via the index
cursor := ""
batchSize := 1000 // Process in batches
for {
records, nextCursor, err := p.recordsIndex.ListRecords(atproto.LayerCollection, batchSize, cursor, true)
if err != nil {
return nil, fmt.Errorf("failed to list layer records: %w", err)
}
for _, rec := range records {
// Construct record path and get the record data
recordPath := rec.Collection + "/" + rec.Rkey
_, recBytes, err := repoHandle.GetRecordBytes(ctx, recordPath)
if err != nil {
// Skip records we can't read
continue
}
// Decode the layer record
recordValue, err := lexutil.CborDecodeValue(*recBytes)
if err != nil {
continue
}
layerRecord, ok := recordValue.(*atproto.LayerRecord)
if !ok {
continue
}
// Filter by userDID
if layerRecord.UserDID != userDID {
continue
}
// Deduplicate by digest - keep the size (could be different pushes of same blob)
// Store the size - we only count each unique digest once
if _, exists := digestSizes[layerRecord.Digest]; !exists {
digestSizes[layerRecord.Digest] = layerRecord.Size
}
}
if nextCursor == "" {
break
}
cursor = nextCursor
}
// Calculate totals
var totalSize int64
for _, size := range digestSizes {
totalSize += size
}
return &QuotaStats{
UserDID: userDID,
UniqueBlobs: len(digestSizes),
TotalSize: totalSize,
}, nil
}
pkg/hold/pds/layer_test.go
+62 -73
View File
@@ -22,9 +22,8 @@ func TestCreateLayerRecord(t *testing.T) {
"sha256:abc123def456",
1048576, // 1 MB
"application/vnd.oci.image.layer.v1.tar+gzip",
"myapp",
"did:plc:alice123",
"alice.bsky.social",
"at://did:plc:alice123/io.atcr.manifest/abc123def456",
),
wantErr: false,
},
@@ -34,22 +33,20 @@ func TestCreateLayerRecord(t *testing.T) {
"sha256:fedcba987654",
1073741824, // 1 GB
"application/vnd.docker.image.rootfs.diff.tar.gzip",
"debian",
"did:plc:bob456",
"bob.example.com",
"at://did:plc:bob456/io.atcr.manifest/fedcba987654",
),
wantErr: false,
},
{
name: "invalid record type",
record: &atproto.LayerRecord{
Type: "wrong.type",
Digest: "sha256:abc123",
Size: 1024,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Repository: "test",
UserDID: "did:plc:test",
UserHandle: "test.example.com",
Type: "wrong.type",
Digest: "sha256:abc123",
Size: 1024,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Manifest: "at://did:plc:test/io.atcr.manifest/abc123",
UserDID: "did:plc:test",
},
wantErr: true,
errSubstr: "invalid record type",
@@ -57,13 +54,12 @@ func TestCreateLayerRecord(t *testing.T) {
{
name: "missing digest",
record: &atproto.LayerRecord{
Type: atproto.LayerCollection,
Digest: "",
Size: 1024,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Repository: "test",
UserDID: "did:plc:test",
UserHandle: "test.example.com",
Type: atproto.LayerCollection,
Digest: "",
Size: 1024,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Manifest: "at://did:plc:test/io.atcr.manifest/abc123",
UserDID: "did:plc:test",
},
wantErr: true,
errSubstr: "digest is required",
@@ -71,13 +67,12 @@ func TestCreateLayerRecord(t *testing.T) {
{
name: "zero size",
record: &atproto.LayerRecord{
Type: atproto.LayerCollection,
Digest: "sha256:abc123",
Size: 0,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Repository: "test",
UserDID: "did:plc:test",
UserHandle: "test.example.com",
Type: atproto.LayerCollection,
Digest: "sha256:abc123",
Size: 0,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Manifest: "at://did:plc:test/io.atcr.manifest/abc123",
UserDID: "did:plc:test",
},
wantErr: true,
errSubstr: "size must be positive",
@@ -85,13 +80,12 @@ func TestCreateLayerRecord(t *testing.T) {
{
name: "negative size",
record: &atproto.LayerRecord{
Type: atproto.LayerCollection,
Digest: "sha256:abc123",
Size: -1,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Repository: "test",
UserDID: "did:plc:test",
UserHandle: "test.example.com",
Type: atproto.LayerCollection,
Digest: "sha256:abc123",
Size: -1,
MediaType: "application/vnd.oci.image.layer.v1.tar",
Manifest: "at://did:plc:test/io.atcr.manifest/abc123",
UserDID: "did:plc:test",
},
wantErr: true,
errSubstr: "size must be positive",
@@ -135,6 +129,8 @@ func TestCreateLayerRecord_MultipleRecords(t *testing.T) {
// Test creating multiple layer records for the same manifest
pds, ctx := setupTestPDS(t)
manifestURI := "at://did:plc:test123/io.atcr.manifest/manifestabc123"
layers := []struct {
digest string
size int64
@@ -151,9 +147,8 @@ func TestCreateLayerRecord_MultipleRecords(t *testing.T) {
layer.digest,
layer.size,
"application/vnd.oci.image.layer.v1.tar+gzip",
"multi-layer-app",
"did:plc:test123",
"test.example.com",
manifestURI,
)
rkey, cid, err := pds.CreateLayerRecord(ctx, record)
@@ -180,11 +175,10 @@ func TestNewLayerRecord(t *testing.T) {
digest := "sha256:abc123def456"
size := int64(1048576)
mediaType := "application/vnd.oci.image.layer.v1.tar+gzip"
repository := "myapp"
userDID := "did:plc:alice123"
userHandle := "alice.bsky.social"
manifestURI := "at://did:plc:alice123/io.atcr.manifest/abc123def456"
record := atproto.NewLayerRecord(digest, size, mediaType, repository, userDID, userHandle)
record := atproto.NewLayerRecord(digest, size, mediaType, userDID, manifestURI)
if record == nil {
t.Fatal("NewLayerRecord() returned nil")
@@ -207,18 +201,14 @@ func TestNewLayerRecord(t *testing.T) {
t.Errorf("MediaType = %q, want %q", record.MediaType, mediaType)
}
if record.Repository != repository {
t.Errorf("Repository = %q, want %q", record.Repository, repository)
if record.Manifest != manifestURI {
t.Errorf("Manifest = %q, want %q", record.Manifest, manifestURI)
}
if record.UserDID != userDID {
t.Errorf("UserDID = %q, want %q", record.UserDID, userDID)
}
if record.UserHandle != userHandle {
t.Errorf("UserHandle = %q, want %q", record.UserHandle, userHandle)
}
if record.CreatedAt == "" {
t.Error("CreatedAt is empty")
}
@@ -229,40 +219,36 @@ func TestNewLayerRecord(t *testing.T) {
func TestLayerRecord_FieldValidation(t *testing.T) {
// Test various field values
tests := []struct {
name string
digest string
size int64
mediaType string
repository string
userDID string
userHandle string
name string
digest string
size int64
mediaType string
userDID string
manifestURI string
}{
{
name: "typical OCI layer",
digest: "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f",
size: 12582912, // 12 MB
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
repository: "hsm-secrets-operator",
userDID: "did:plc:evan123",
userHandle: "evan.jarrett.net",
name: "typical OCI layer",
digest: "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f",
size: 12582912, // 12 MB
mediaType: "application/vnd.oci.image.layer.v1.tar+gzip",
userDID: "did:plc:evan123",
manifestURI: "at://did:plc:evan123/io.atcr.manifest/abc123",
},
{
name: "Docker layer format",
digest: "sha256:abc123",
size: 1024,
mediaType: "application/vnd.docker.image.rootfs.diff.tar.gzip",
repository: "nginx",
userDID: "did:plc:user456",
userHandle: "user.example.com",
name: "Docker layer format",
digest: "sha256:abc123",
size: 1024,
mediaType: "application/vnd.docker.image.rootfs.diff.tar.gzip",
userDID: "did:plc:user456",
manifestURI: "at://did:plc:user456/io.atcr.manifest/def456",
},
{
name: "uncompressed layer",
digest: "sha256:def456",
size: 2048,
mediaType: "application/vnd.oci.image.layer.v1.tar",
repository: "alpine",
userDID: "did:plc:user789",
userHandle: "user.bsky.social",
name: "uncompressed layer",
digest: "sha256:def456",
size: 2048,
mediaType: "application/vnd.oci.image.layer.v1.tar",
userDID: "did:plc:user789",
manifestURI: "at://did:plc:user789/io.atcr.manifest/ghi789",
},
}
@@ -272,9 +258,8 @@ func TestLayerRecord_FieldValidation(t *testing.T) {
tt.digest,
tt.size,
tt.mediaType,
tt.repository,
tt.userDID,
tt.userHandle,
tt.manifestURI,
)
if record == nil {
@@ -289,6 +274,10 @@ func TestLayerRecord_FieldValidation(t *testing.T) {
if record.Digest != tt.digest {
t.Errorf("Digest = %q, want %q", record.Digest, tt.digest)
}
if record.Manifest != tt.manifestURI {
t.Errorf("Manifest = %q, want %q", record.Manifest, tt.manifestURI)
}
})
}
}
pkg/hold/pds/xrpc.go
+31
View File
@@ -192,6 +192,9 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) {
r.Use(h.requireAuth)
r.Post(atproto.HoldRequestCrew, h.HandleRequestCrew)
})
// Public quota endpoint (no auth - quota is per-user, just needs userDid param)
r.Get(atproto.HoldGetQuota, h.HandleGetQuota)
}
// HandleHealth returns health check information
@@ -1513,3 +1516,31 @@ func getProxyURL(publicURL string, digest, did string, operation string) string
// Clients should use multipart upload flow via com.atproto.repo.uploadBlob
return ""
}
// HandleGetQuota returns storage quota information for a user
// This calculates the total unique blob storage used by a specific user
// by iterating layer records and deduplicating by digest.
func (h *XRPCHandler) HandleGetQuota(w http.ResponseWriter, r *http.Request) {
userDID := r.URL.Query().Get("userDid")
if userDID == "" {
http.Error(w, "missing required parameter: userDid", http.StatusBadRequest)
return
}
// Validate DID format
if !atproto.IsDID(userDID) {
http.Error(w, "invalid userDid format", http.StatusBadRequest)
return
}
// Get quota stats
stats, err := h.pds.GetQuotaForUser(r.Context(), userDID)
if err != nil {
slog.Error("Failed to get quota", "userDid", userDID, "error", err)
http.Error(w, fmt.Sprintf("failed to get quota: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(stats)
}