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

more jetstream and ui improvements

Browse Source
This commit is contained in:
Evan Jarrett
2025-10-06 17:39:21 -05:00
parent 383face791
commit e56634d6c8
17 changed files with 696 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+2 -2
View File
@@ -22,8 +22,8 @@ RUN CGO_ENABLED=1 GOOS=linux go build -a -o atcr-registry ./cmd/registry
# Runtime stage
FROM alpine:latest
# Install CA certificates for HTTPS and SQLite runtime libraries
RUN apk --no-cache add ca-certificates sqlite-libs
# Install CA certificates for HTTPS, SQLite runtime libraries, and sqlite CLI for debugging
RUN apk --no-cache add ca-certificates sqlite-libs sqlite
# Set working directory
WORKDIR /app
cmd/registry/serve.go
+60 -11
View File
@@ -27,6 +27,7 @@ import (
"atcr.io/pkg/appview"
"atcr.io/pkg/appview/db"
uihandlers "atcr.io/pkg/appview/handlers"
"atcr.io/pkg/appview/jetstream"
appmiddleware "atcr.io/pkg/appview/middleware"
appsession "atcr.io/pkg/appview/session"
"github.com/gorilla/mux"
@@ -127,7 +128,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error {
middleware.SetGlobalRefresher(refresher)
// 6. Initialize UI components (get session store for OAuth integration)
uiDatabase, uiSessionStore, uiTemplates, uiRouter := initializeUI(config)
uiDatabase, uiSessionStore, uiTemplates, uiRouter := initializeUI(config, refresher, baseURL)
// 7. Create OAuth server
oauthServer := oauth.NewServer(refreshStorage, sessionManager, baseURL)
@@ -338,7 +339,7 @@ func extractDefaultHoldEndpoint(config *configuration.Configuration) string {
}
// initializeUI initializes the web UI components
func initializeUI(config *configuration.Configuration) (*sql.DB, *appsession.Store, *template.Template, *mux.Router) {
func initializeUI(config *configuration.Configuration, refresher *oauth.Refresher, baseURL string) (*sql.DB, *appsession.Store, *template.Template, *mux.Router) {
// Check if UI is enabled (optional configuration)
uiEnabled := os.Getenv("ATCR_UI_ENABLED")
if uiEnabled == "false" {
@@ -367,8 +368,12 @@ func initializeUI(config *configuration.Configuration) (*sql.DB, *appsession.Sto
fmt.Printf("UI database initialized at %s\n", dbPath)
// Create session store
sessionStore := appsession.NewStore()
// Create session store with file persistence
sessionStorePath := os.Getenv("ATCR_UI_SESSION_PATH")
if sessionStorePath == "" {
sessionStorePath = "/var/lib/atcr/ui-sessions.json"
}
sessionStore := appsession.NewStore(sessionStorePath)
// Start cleanup goroutine
go func() {
@@ -399,15 +404,17 @@ func initializeUI(config *configuration.Configuration) (*sql.DB, *appsession.Sto
// Public routes (with optional auth for navbar)
router.Handle("/", appmiddleware.OptionalAuth(sessionStore)(
&uihandlers.HomeHandler{
DB: database,
Templates: templates,
DB: database,
Templates: templates,
RegistryURL: baseURL,
},
)).Methods("GET")
router.Handle("/api/recent-pushes", appmiddleware.OptionalAuth(sessionStore)(
&uihandlers.RecentPushesHandler{
DB: database,
Templates: templates,
DB: database,
Templates: templates,
RegistryURL: baseURL,
},
)).Methods("GET")
@@ -416,15 +423,19 @@ func initializeUI(config *configuration.Configuration) (*sql.DB, *appsession.Sto
authRouter.Use(appmiddleware.RequireAuth(sessionStore))
authRouter.Handle("/images", &uihandlers.ImagesHandler{
DB: database,
Templates: templates,
DB: database,
Templates: templates,
RegistryURL: baseURL,
}).Methods("GET")
authRouter.Handle("/settings", &uihandlers.SettingsHandler{
Templates: templates,
Refresher: refresher,
}).Methods("GET")
authRouter.Handle("/api/profile/default-hold", &uihandlers.UpdateDefaultHoldHandler{}).Methods("POST")
authRouter.Handle("/api/profile/default-hold", &uihandlers.UpdateDefaultHoldHandler{
Refresher: refresher,
}).Methods("POST")
authRouter.Handle("/api/images/{repository}/tags/{tag}", &uihandlers.DeleteTagHandler{
DB: database,
@@ -443,5 +454,43 @@ func initializeUI(config *configuration.Configuration) (*sql.DB, *appsession.Sto
http.Redirect(w, r, "/", http.StatusFound)
}).Methods("POST")
// Start Jetstream worker
jetstreamURL := os.Getenv("JETSTREAM_URL")
if jetstreamURL == "" {
jetstreamURL = "wss://jetstream2.us-west.bsky.network/subscribe"
}
// Parse cursor for backfilling historical data
// Set to Unix microseconds timestamp to replay from that point
// Examples:
// - 2 weeks ago: use `date -d '2 weeks ago' +%s` * 1000000
// - Leave unset (or 0) to start from now
var startCursor int64
if cursorStr := os.Getenv("JETSTREAM_START_CURSOR"); cursorStr != "" {
if cursor, err := time.Parse(time.RFC3339, cursorStr); err == nil {
// Support RFC3339 format: "2025-09-23T00:00:00Z"
startCursor = cursor.UnixMicro()
fmt.Printf("Jetstream: Starting from %s (%d microseconds)\n", cursorStr, startCursor)
} else if cursor, err := time.ParseDuration(cursorStr); err == nil {
// Support duration format: "-336h" (2 weeks ago)
startCursor = time.Now().Add(cursor).UnixMicro()
fmt.Printf("Jetstream: Starting from %s ago (%d microseconds)\n", cursorStr, startCursor)
} else {
fmt.Printf("Warning: Invalid JETSTREAM_START_CURSOR format: %s\n", cursorStr)
}
}
worker := jetstream.NewWorker(database, jetstreamURL, startCursor)
go func() {
for {
if err := worker.Start(context.Background()); err != nil {
fmt.Printf("Jetstream worker error: %v, reconnecting in 10s...\n", err)
time.Sleep(10 * time.Second)
}
}
}()
fmt.Println("Jetstream worker started")
return database, sessionStore, templates, router
}
docker-compose.yml
+2
View File
@@ -10,6 +10,8 @@ services:
environment:
- ATCR_TOKEN_STORAGE_PATH=/var/lib/atcr/tokens/oauth-tokens.json
- ATCR_UI_ENABLED=true
# Jetstream backfill: Replay 5 days of historical events
# - JETSTREAM_START_CURSOR=-120h
volumes:
# Auth keys (JWT signing keys)
- atcr-auth:/var/lib/atcr/auth
go.mod
+2 -1
View File
@@ -30,12 +30,13 @@ require (
github.com/go-logr/stdr v1.2.2 // indirect
github.com/gorilla/handlers v1.5.2 // indirect
github.com/gorilla/mux v1.8.1 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0 // indirect
github.com/hashicorp/golang-lru/arc/v2 v2.0.6 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/klauspost/compress v1.17.11 // indirect
github.com/klauspost/compress v1.18.0 // indirect
github.com/mattn/go-sqlite3 v1.14.32 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/opencontainers/image-spec v1.1.0 // indirect
go.sum
+4
View File
@@ -77,6 +77,8 @@ github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyE
github.com/gorilla/handlers v1.5.2/go.mod h1:dX+xVpaxdSw+q0Qek8SSsl3dfMk3jNddUkMzo0GtH0w=
github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0 h1:ad0vkEBuk23VJzZR9nkLVG0YAoN9coASF1GusYX6AlU=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0/go.mod h1:igFoXX2ELCW06bol23DWPB5BEWfZISOzSP5K2sbLea0=
github.com/hashicorp/golang-lru/arc/v2 v2.0.6 h1:4NU7uP5vSoK6TbaMj3NtY478TTAWLso/vL1gpNrInHg=
@@ -94,6 +96,8 @@ github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/u
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
pkg/appview/db/queries.go
+22 -1
View File
@@ -2,6 +2,7 @@ package db
import (
"database/sql"
"time"
)
// GetRecentPushes fetches recent pushes with pagination
@@ -81,10 +82,30 @@ func GetUserRepositories(db *sql.DB, did string) ([]Repository, error) {
var repos []Repository
for rows.Next() {
var r Repository
if err := rows.Scan(&r.Name, &r.TagCount, &r.ManifestCount, &r.LastPush); err != nil {
var lastPushStr string
if err := rows.Scan(&r.Name, &r.TagCount, &r.ManifestCount, &lastPushStr); err != nil {
return nil, err
}
// Parse the timestamp string into time.Time
if lastPushStr != "" {
// Try multiple timestamp formats
formats := []string{
time.RFC3339Nano, // 2006-01-02T15:04:05.999999999Z07:00
"2006-01-02 15:04:05.999999999-07:00", // SQLite with microseconds and timezone
"2006-01-02 15:04:05.999999999", // SQLite with microseconds
time.RFC3339, // 2006-01-02T15:04:05Z07:00
"2006-01-02 15:04:05", // SQLite default
}
for _, format := range formats {
if t, err := time.Parse(format, lastPushStr); err == nil {
r.LastPush = t
break
}
}
}
// Get tags for this repo
tagRows, err := db.Query(`
SELECT id, tag, digest, created_at
pkg/appview/handlers/home.go
+20 -14
View File
@@ -12,17 +12,20 @@ import (
// HomeHandler handles the home page
type HomeHandler struct {
DB *sql.DB
Templates *template.Template
DB *sql.DB
Templates *template.Template
RegistryURL string
}
func (h *HomeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
data := struct {
User *db.User
Query string
User *db.User
Query string
RegistryURL string
}{
User: middleware.GetUser(r),
Query: r.URL.Query().Get("q"),
User: middleware.GetUser(r),
Query: r.URL.Query().Get("q"),
RegistryURL: h.RegistryURL,
}
if err := h.Templates.ExecuteTemplate(w, "home", data); err != nil {
@@ -33,8 +36,9 @@ func (h *HomeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// RecentPushesHandler handles the HTMX request for recent pushes
type RecentPushesHandler struct {
DB *sql.DB
Templates *template.Template
DB *sql.DB
Templates *template.Template
RegistryURL string
}
func (h *RecentPushesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -57,13 +61,15 @@ func (h *RecentPushesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
}
data := struct {
Pushes []db.Push
HasMore bool
NextOffset int
Pushes []db.Push
HasMore bool
NextOffset int
RegistryURL string
}{
Pushes: pushes,
HasMore: offset+limit < total,
NextOffset: offset + limit,
Pushes: pushes,
HasMore: offset+limit < total,
NextOffset: offset + limit,
RegistryURL: h.RegistryURL,
}
if err := h.Templates.ExecuteTemplate(w, "push-list.html", data); err != nil {
pkg/appview/handlers/images.go
+5 -2
View File
@@ -12,8 +12,9 @@ import (
// ImagesHandler handles the images management page
type ImagesHandler struct {
DB *sql.DB
Templates *template.Template
DB *sql.DB
Templates *template.Template
RegistryURL string
}
func (h *ImagesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -34,10 +35,12 @@ func (h *ImagesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
User *db.User
Repositories []db.Repository
Query string
RegistryURL string
}{
User: user,
Repositories: repos,
Query: r.URL.Query().Get("q"),
RegistryURL: h.RegistryURL,
}
if err := h.Templates.ExecuteTemplate(w, "images", data); err != nil {
pkg/appview/handlers/settings.go
+53 -9
View File
@@ -1,29 +1,50 @@
package handlers
import (
"fmt"
"html/template"
"net/http"
"time"
"atcr.io/pkg/appview/db"
"atcr.io/pkg/appview/middleware"
"atcr.io/pkg/atproto"
"atcr.io/pkg/auth/oauth"
)
// SettingsHandler handles the settings page
type SettingsHandler struct {
Templates *template.Template
// TODO: Add ATProto client when implementing profile fetching
Refresher *oauth.Refresher
}
func (h *SettingsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
user := middleware.GetUser(r)
if user == nil {
http.Redirect(w, r, "/auth/oauth/login?return_to=/ui/settings", http.StatusFound)
http.Redirect(w, r, "/auth/oauth/login?return_to=/settings", http.StatusFound)
return
}
// TODO: Fetch actual profile from PDS using ATProto client
// For now, using mock data from session
// Get access token and DPoP transport for the user
accessToken, _, dpopTransport, err := h.Refresher.GetAccessToken(r.Context(), user.DID)
if err != nil {
http.Error(w, "Failed to get access token: "+err.Error(), http.StatusInternalServerError)
return
}
// Create ATProto client with DPoP transport
client := atproto.NewClientWithDPoP(user.PDSEndpoint, user.DID, accessToken, nil, dpopTransport)
// Fetch sailor profile
profile, err := atproto.GetProfile(r.Context(), client)
if err != nil {
// Log error but don't fail - profile might not exist yet
fmt.Printf("WARNING [settings]: Failed to fetch profile for %s: %v\n", user.DID, err)
profile = &atproto.SailorProfileRecord{}
} else {
fmt.Printf("DEBUG [settings]: Fetched profile for %s: defaultHold=%s\n", user.DID, profile.DefaultHold)
}
data := struct {
User *db.User
Profile struct {
@@ -43,7 +64,7 @@ func (h *SettingsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
data.Profile.Handle = user.Handle
data.Profile.DID = user.DID
data.Profile.PDSEndpoint = user.PDSEndpoint
// data.Profile.DefaultHold will be empty for now
data.Profile.DefaultHold = profile.DefaultHold
if err := h.Templates.ExecuteTemplate(w, "settings", data); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -53,7 +74,7 @@ func (h *SettingsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// UpdateDefaultHoldHandler handles updating the default hold
type UpdateDefaultHoldHandler struct {
// TODO: Add ATProto client for updating profile
Refresher *oauth.Refresher
}
func (h *UpdateDefaultHoldHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -65,9 +86,32 @@ func (h *UpdateDefaultHoldHandler) ServeHTTP(w http.ResponseWriter, r *http.Requ
holdEndpoint := r.FormValue("hold_endpoint")
// TODO: Update profile in PDS via ATProto client
// For now, just return success
_ = holdEndpoint
// Get access token and DPoP transport for the user
accessToken, _, dpopTransport, err := h.Refresher.GetAccessToken(r.Context(), user.DID)
if err != nil {
http.Error(w, "Failed to get access token: "+err.Error(), http.StatusInternalServerError)
return
}
// Create ATProto client with DPoP transport
client := atproto.NewClientWithDPoP(user.PDSEndpoint, user.DID, accessToken, nil, dpopTransport)
// Fetch existing profile or create new one
profile, err := atproto.GetProfile(r.Context(), client)
if err != nil || profile == nil {
// Profile doesn't exist, create new one
profile = atproto.NewSailorProfileRecord(holdEndpoint)
} else {
// Update existing profile
profile.DefaultHold = holdEndpoint
profile.UpdatedAt = time.Now()
}
// Save profile
if err := atproto.UpdateProfile(r.Context(), client, profile); err != nil {
http.Error(w, "Failed to update profile: "+err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/html")
w.Write([]byte(`<div class="success">✓ Default hold updated successfully!</div>`))
pkg/appview/jetstream/worker.go
+383
View File
@@ -0,0 +1,383 @@
package jetstream
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"net/url"
"strings"
"time"
"atcr.io/pkg/appview/db"
"atcr.io/pkg/atproto"
"github.com/gorilla/websocket"
"github.com/klauspost/compress/zstd"
)
// UserCache caches DID -> handle/PDS mappings to avoid repeated lookups
type UserCache struct {
cache map[string]*db.User
}
// Worker consumes Jetstream events and populates the UI database
type Worker struct {
db *sql.DB
jetstreamURL string
startCursor int64
wantedCollections []string
debugCollectionCount int
userCache *UserCache
resolver *atproto.Resolver
}
// NewWorker creates a new Jetstream worker
// startCursor: Unix microseconds timestamp to start from (0 = start from now)
func NewWorker(database *sql.DB, jetstreamURL string, startCursor int64) *Worker {
if jetstreamURL == "" {
jetstreamURL = "wss://jetstream2.us-west.bsky.network/subscribe"
}
return &Worker{
db: database,
jetstreamURL: jetstreamURL,
startCursor: startCursor,
wantedCollections: []string{
atproto.ManifestCollection, // io.atcr.manifest
atproto.TagCollection, // io.atcr.tag
},
userCache: &UserCache{
cache: make(map[string]*db.User),
},
resolver: atproto.NewResolver(),
}
}
// Start begins consuming Jetstream events
// This is a blocking function that runs until the context is cancelled
func (w *Worker) Start(ctx context.Context) error {
// Build connection URL with filters
u, err := url.Parse(w.jetstreamURL)
if err != nil {
return fmt.Errorf("invalid jetstream URL: %w", err)
}
q := u.Query()
for _, collection := range w.wantedCollections {
q.Add("wantedCollections", collection)
}
// Add cursor if specified (for backfilling historical data)
if w.startCursor > 0 {
q.Set("cursor", fmt.Sprintf("%d", w.startCursor))
fmt.Printf("Starting from cursor: %d (replaying historical events)\n", w.startCursor)
}
// Disable compression for now to debug
// q.Set("compress", "true")
u.RawQuery = q.Encode()
fmt.Printf("Connecting to Jetstream: %s\n", u.String())
// Connect to Jetstream
conn, _, err := websocket.DefaultDialer.DialContext(ctx, u.String(), nil)
if err != nil {
return fmt.Errorf("failed to connect to jetstream: %w", err)
}
defer conn.Close()
// Create zstd decoder for decompressing messages
decoder, err := zstd.NewReader(nil)
if err != nil {
return fmt.Errorf("failed to create zstd decoder: %w", err)
}
defer decoder.Close()
fmt.Println("Connected to Jetstream, listening for events...")
// Start heartbeat ticker to show Jetstream is alive
heartbeatTicker := time.NewTicker(30 * time.Second)
defer heartbeatTicker.Stop()
eventCount := 0
lastHeartbeat := time.Now()
// Read messages
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-heartbeatTicker.C:
elapsed := time.Since(lastHeartbeat)
fmt.Printf("Jetstream: Alive (processed %d events in last %.0fs)\n", eventCount, elapsed.Seconds())
eventCount = 0
lastHeartbeat = time.Now()
default:
_, message, err := conn.ReadMessage()
if err != nil {
return fmt.Errorf("failed to read message: %w", err)
}
// For now, process uncompressed messages
// TODO: Re-enable compression once debugging is complete
_ = decoder // Keep decoder to avoid unused variable error
if err := w.processMessage(message); err != nil {
fmt.Printf("ERROR processing message: %v\n", err)
// Continue processing other messages
} else {
eventCount++
}
}
}
}
// processMessage processes a single Jetstream event
func (w *Worker) processMessage(message []byte) error {
var event JetstreamEvent
if err := json.Unmarshal(message, &event); err != nil {
return fmt.Errorf("failed to unmarshal event: %w", err)
}
// Only process commit events
if event.Kind != "commit" {
return nil
}
commit := event.Commit
if commit == nil {
return nil
}
// Set DID on commit from parent event
commit.DID = event.DID
// Debug: log first few collections we see to understand what's coming through
if w.debugCollectionCount < 5 {
fmt.Printf("Jetstream DEBUG: Received collection=%s, did=%s\n", commit.Collection, commit.DID)
w.debugCollectionCount++
}
// Process based on collection
switch commit.Collection {
case atproto.ManifestCollection:
fmt.Printf("Jetstream: Processing manifest event: did=%s, operation=%s, rkey=%s\n",
commit.DID, commit.Operation, commit.RKey)
return w.processManifest(commit)
case atproto.TagCollection:
fmt.Printf("Jetstream: Processing tag event: did=%s, operation=%s, rkey=%s\n",
commit.DID, commit.Operation, commit.RKey)
return w.processTag(commit)
default:
// Ignore other collections
return nil
}
}
// ensureUser resolves and upserts a user by DID
func (w *Worker) ensureUser(ctx context.Context, did string) error {
// Check cache first
if user, ok := w.userCache.cache[did]; ok {
// Update last seen
user.LastSeen = time.Now()
return db.UpsertUser(w.db, user)
}
// Resolve DID to get handle and PDS endpoint
resolvedDID, pdsEndpoint, err := w.resolver.ResolveIdentity(ctx, did)
if err != nil {
fmt.Printf("WARNING: Failed to resolve DID %s: %v (using DID as handle)\n", did, err)
// Fallback: use DID as handle
resolvedDID = did
pdsEndpoint = "https://bsky.social" // Default PDS endpoint as fallback
}
// Get handle from DID document
handle, err := w.resolver.ResolveHandleFromDID(ctx, resolvedDID)
if err != nil {
fmt.Printf("WARNING: Failed to get handle for DID %s: %v (using DID as handle)\n", resolvedDID, err)
handle = resolvedDID // Fallback to DID
}
// Cache the user
user := &db.User{
DID: resolvedDID,
Handle: handle,
PDSEndpoint: pdsEndpoint,
LastSeen: time.Now(),
}
w.userCache.cache[did] = user
// Upsert to database
return db.UpsertUser(w.db, user)
}
// processManifest processes a manifest commit event
func (w *Worker) processManifest(commit *CommitEvent) error {
// Resolve and upsert user with handle/PDS endpoint
if err := w.ensureUser(context.Background(), commit.DID); err != nil {
return fmt.Errorf("failed to ensure user: %w", err)
}
if commit.Operation == "delete" {
// Delete manifest
repo := extractRepoFromRKey(commit.RKey)
digest := commit.RKey
return db.DeleteManifest(w.db, commit.DID, repo, digest)
}
// Parse manifest record
var manifestRecord atproto.ManifestRecord
if commit.Record != nil {
recordBytes, err := json.Marshal(commit.Record)
if err != nil {
return fmt.Errorf("failed to marshal record: %w", err)
}
if err := json.Unmarshal(recordBytes, &manifestRecord); err != nil {
return fmt.Errorf("failed to unmarshal manifest: %w", err)
}
} else {
// No record data, can't process
return nil
}
// Serialize full manifest as JSON for storage
manifestJSON, err := json.Marshal(manifestRecord)
if err != nil {
return fmt.Errorf("failed to marshal manifest: %w", err)
}
// Insert manifest
manifestID, err := db.InsertManifest(w.db, &db.Manifest{
DID: commit.DID,
Repository: manifestRecord.Repository,
Digest: manifestRecord.Digest,
MediaType: manifestRecord.MediaType,
SchemaVersion: manifestRecord.SchemaVersion,
ConfigDigest: manifestRecord.Config.Digest,
ConfigSize: manifestRecord.Config.Size,
RawManifest: string(manifestJSON),
HoldEndpoint: manifestRecord.HoldEndpoint,
CreatedAt: manifestRecord.CreatedAt,
})
if err != nil {
return fmt.Errorf("failed to insert manifest: %w", err)
}
// Insert layers
for i, layer := range manifestRecord.Layers {
if err := db.InsertLayer(w.db, &db.Layer{
ManifestID: manifestID,
Digest: layer.Digest,
MediaType: layer.MediaType,
Size: layer.Size,
LayerIndex: i,
}); err != nil {
// Continue on error - layer might already exist
continue
}
}
return nil
}
// processTag processes a tag commit event
func (w *Worker) processTag(commit *CommitEvent) error {
// Resolve and upsert user with handle/PDS endpoint
if err := w.ensureUser(context.Background(), commit.DID); err != nil {
return fmt.Errorf("failed to ensure user: %w", err)
}
if commit.Operation == "delete" {
// Delete tag
parts := strings.Split(commit.RKey, "/")
if len(parts) < 2 {
return fmt.Errorf("invalid tag rkey: %s", commit.RKey)
}
repo := strings.Join(parts[:len(parts)-1], "/")
tag := parts[len(parts)-1]
return db.DeleteTag(w.db, commit.DID, repo, tag)
}
// Parse tag record
var tagRecord atproto.TagRecord
if commit.Record != nil {
recordBytes, err := json.Marshal(commit.Record)
if err != nil {
return fmt.Errorf("failed to marshal record: %w", err)
}
if err := json.Unmarshal(recordBytes, &tagRecord); err != nil {
return fmt.Errorf("failed to unmarshal tag: %w", err)
}
} else {
return nil
}
// Insert or update tag
return db.UpsertTag(w.db, &db.Tag{
DID: commit.DID,
Repository: tagRecord.Repository,
Tag: tagRecord.Tag,
Digest: tagRecord.ManifestDigest,
CreatedAt: tagRecord.UpdatedAt,
})
}
// JetstreamEvent represents a Jetstream event
type JetstreamEvent struct {
DID string `json:"did"`
TimeUS int64 `json:"time_us"`
Kind string `json:"kind"` // "commit", "identity", "account"
Commit *CommitEvent `json:"commit,omitempty"`
Identity *IdentityInfo `json:"identity,omitempty"`
Account *AccountInfo `json:"account,omitempty"`
}
// CommitEvent represents a commit event (create/update/delete)
type CommitEvent struct {
Rev string `json:"rev"`
Operation string `json:"operation"` // "create", "update", "delete"
Collection string `json:"collection"`
RKey string `json:"rkey"`
Record map[string]interface{} `json:"record,omitempty"`
CID string `json:"cid,omitempty"`
DID string `json:"-"` // Set from parent event
}
// IdentityInfo represents an identity event
type IdentityInfo struct {
DID string `json:"did"`
Handle string `json:"handle"`
Seq int64 `json:"seq"`
Time string `json:"time"`
}
// AccountInfo represents an account status event
type AccountInfo struct {
Active bool `json:"active"`
DID string `json:"did"`
Seq int64 `json:"seq"`
Time string `json:"time"`
Status string `json:"status,omitempty"`
}
// Helper functions
func extractRepoFromRKey(rkey string) string {
// RKey format: <digest> or <repo>/<digest>
// For manifest, it's just the digest
parts := strings.Split(rkey, "/")
if len(parts) > 1 {
return parts[0]
}
return ""
}
func calculateManifestSize(manifest *atproto.ManifestRecord) int64 {
var total int64
total += manifest.Config.Size
for _, layer := range manifest.Layers {
total += layer.Size
}
return total
}
pkg/appview/middleware/auth.go
+6 -4
View File
@@ -29,8 +29,9 @@ func RequireAuth(store *session.Store) func(http.Handler) http.Handler {
}
user := &db.User{
DID: sess.DID,
Handle: sess.Handle,
DID: sess.DID,
Handle: sess.Handle,
PDSEndpoint: sess.PDSEndpoint,
}
ctx := context.WithValue(r.Context(), userKey, user)
@@ -47,8 +48,9 @@ func OptionalAuth(store *session.Store) func(http.Handler) http.Handler {
if ok {
if sess, ok := store.Get(sessionID); ok {
user := &db.User{
DID: sess.DID,
Handle: sess.Handle,
DID: sess.DID,
Handle: sess.Handle,
PDSEndpoint: sess.PDSEndpoint,
}
ctx := context.WithValue(r.Context(), userKey, user)
r = r.WithContext(ctx)
pkg/appview/session/session.go
+92 -13
View File
@@ -3,34 +3,92 @@ package session
import (
"crypto/rand"
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"os"
"sync"
"time"
)
// Session represents a user session
type Session struct {
ID string
DID string
Handle string
ExpiresAt time.Time
ID string
DID string
Handle string
PDSEndpoint string
ExpiresAt time.Time
}
// Store manages user sessions
type Store struct {
mu sync.RWMutex
sessions map[string]*Session
filePath string
}
// NewStore creates a new session store
func NewStore() *Store {
return &Store{
// NewStore creates a new session store with file persistence
func NewStore(filePath string) *Store {
store := &Store{
sessions: make(map[string]*Session),
filePath: filePath,
}
// Load existing sessions from file
if err := store.load(); err != nil {
fmt.Printf("Warning: Failed to load sessions from %s: %v\n", filePath, err)
}
return store
}
// Create creates a new session and returns the full Session struct
func (s *Store) Create(did, handle string, duration time.Duration) (string, error) {
// load reads sessions from disk
func (s *Store) load() error {
if s.filePath == "" {
return nil
}
data, err := os.ReadFile(s.filePath)
if err != nil {
if os.IsNotExist(err) {
return nil // File doesn't exist yet, that's fine
}
return err
}
var sessions map[string]*Session
if err := json.Unmarshal(data, &sessions); err != nil {
return err
}
// Filter out expired sessions
now := time.Now()
for id, sess := range sessions {
if now.Before(sess.ExpiresAt) {
s.sessions[id] = sess
}
}
fmt.Printf("Loaded %d active sessions from disk\n", len(s.sessions))
return nil
}
// save writes sessions to disk
func (s *Store) save() error {
if s.filePath == "" {
return nil
}
data, err := json.Marshal(s.sessions)
if err != nil {
return err
}
return os.WriteFile(s.filePath, data, 0600)
}
// Create creates a new session and returns the session ID
func (s *Store) Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
@@ -41,13 +99,20 @@ func (s *Store) Create(did, handle string, duration time.Duration) (string, erro
}
sess := &Session{
ID: base64.URLEncoding.EncodeToString(b),
DID: did,
Handle: handle,
ExpiresAt: time.Now().Add(duration),
ID: base64.URLEncoding.EncodeToString(b),
DID: did,
Handle: handle,
PDSEndpoint: pdsEndpoint,
ExpiresAt: time.Now().Add(duration),
}
s.sessions[sess.ID] = sess
// Save to disk
if err := s.save(); err != nil {
fmt.Printf("Warning: Failed to save sessions to disk: %v\n", err)
}
return sess.ID, nil
}
@@ -70,6 +135,11 @@ func (s *Store) Delete(id string) {
defer s.mu.Unlock()
delete(s.sessions, id)
// Save to disk
if err := s.save(); err != nil {
fmt.Printf("Warning: Failed to save sessions to disk: %v\n", err)
}
}
// Cleanup removes expired sessions
@@ -78,9 +148,18 @@ func (s *Store) Cleanup() {
defer s.mu.Unlock()
now := time.Now()
deleted := 0
for id, sess := range s.sessions {
if now.After(sess.ExpiresAt) {
delete(s.sessions, id)
deleted++
}
}
if deleted > 0 {
// Save to disk
if err := s.save(); err != nil {
fmt.Printf("Warning: Failed to save sessions to disk: %v\n", err)
}
}
}
pkg/appview/static/js/app.js
+14
View File
@@ -58,3 +58,17 @@ document.addEventListener('htmx:afterSwap', updateTimestamps);
// Update timestamps periodically
setInterval(updateTimestamps, 60000); // Every minute
// Toggle repository details (for images page)
function toggleRepo(name) {
const details = document.getElementById('repo-' + name);
const btn = document.getElementById('btn-' + name);
if (details.style.display === 'none') {
details.style.display = 'block';
btn.textContent = '▲';
} else {
details.style.display = 'none';
btn.textContent = '▼';
}
}
pkg/appview/templates/pages/images.html
+10 -25
View File
@@ -7,6 +7,7 @@
<title>Your Images - ATCR</title>
<link rel="stylesheet" href="/static/css/style.css">
<script src="https://unpkg.com/htmx.org@1.9.10"></script>
<script src="/static/js/app.js"></script>
</head>
<body>
{{ template "nav" . }}
@@ -17,10 +18,11 @@
{{ if .Repositories }}
{{ range .Repositories }}
{{ $repoName := .Name }}
<div class="repository-card">
<div class="repo-header" onclick="toggleRepo('{{ .Name }}')">
<div class="repo-header" onclick="toggleRepo('{{ $repoName }}')">
<div>
<h2>{{ .Name }}</h2>
<h2>{{ $repoName }}</h2>
<div class="repo-stats">
<span>{{ .TagCount }} tags</span>
<span>•</span>
@@ -31,16 +33,16 @@
</time>
</div>
</div>
<button class="expand-btn" id="btn-{{ .Name }}">▼</button>
<button class="expand-btn" id="btn-{{ $repoName }}">▼</button>
</div>
<div id="repo-{{ .Name }}" class="repo-details" style="display: none;">
<div id="repo-{{ $repoName }}" class="repo-details" style="display: none;">
<!-- Tags Section -->
<div class="tags-section">
<h3>Tags</h3>
{{ if .Tags }}
{{ range .Tags }}
<div class="tag-row" id="tag-{{ $.Name }}-{{ .Tag }}">
<div class="tag-row" id="tag-{{ $repoName }}-{{ .Tag }}">
<span class="tag-name">{{ .Tag }}</span>
<span class="tag-arrow">→</span>
<code class="tag-digest">{{ truncateDigest .Digest 12 }}</code>
@@ -49,9 +51,9 @@
</time>
<button class="delete-btn"
hx-delete="/api/images/{{ $.Name }}/tags/{{ .Tag }}"
hx-delete="/api/images/{{ $repoName }}/tags/{{ .Tag }}"
hx-confirm="Delete tag {{ .Tag }}?"
hx-target="#tag-{{ $.Name }}-{{ .Tag }}"
hx-target="#tag-{{ $repoName }}-{{ .Tag }}"
hx-swap="outerHTML">
🗑️
</button>
@@ -85,7 +87,7 @@
{{ else }}
<div class="empty-state">
<p>No images yet. Push your first image:</p>
<pre><code>docker push atcr.io/{{ .User.Handle }}/myapp:latest</code></pre>
<pre><code>docker push {{ .RegistryURL }}/{{ .User.Handle }}/myapp:latest</code></pre>
</div>
{{ end }}
</div>
@@ -93,23 +95,6 @@
<!-- Modal container for HTMX -->
<div id="modal"></div>
<script src="/static/js/app.js"></script>
<script>
// Toggle repository details
function toggleRepo(name) {
const details = document.getElementById('repo-' + name);
const btn = document.getElementById('btn-' + name);
if (details.style.display === 'none') {
details.style.display = 'block';
btn.textContent = '▲';
} else {
details.style.display = 'none';
btn.textContent = '▼';
}
}
</script>
</body>
</html>
{{ end }}
pkg/appview/templates/partials/push-list.html
+3 -3
View File
@@ -19,8 +19,8 @@
</div>
<div class="push-command">
<code class="pull-command">docker pull atcr.io/{{ .Handle }}/{{ .Repository }}:{{ .Tag }}</code>
<button class="copy-btn" onclick="copyToClipboard('docker pull atcr.io/{{ .Handle }}/{{ .Repository }}:{{ .Tag }}')">
<code class="pull-command">docker pull {{ $.RegistryURL }}/{{ .Handle }}/{{ .Repository }}:{{ .Tag }}</code>
<button class="copy-btn" onclick="copyToClipboard('docker pull {{ $.RegistryURL }}/{{ .Handle }}/{{ .Repository }}:{{ .Tag }}')">
📋 Copy
</button>
</div>
@@ -39,6 +39,6 @@
{{ if eq (len .Pushes) 0 }}
<div class="empty-state">
<p>No pushes yet. Start using ATCR by pushing your first image!</p>
<pre><code>docker push atcr.io/yourhandle/myapp:latest</code></pre>
<pre><code>docker push {{ .RegistryURL }}/yourhandle/myapp:latest</code></pre>
</div>
{{ end }}
pkg/auth/oauth/refresher.go
+15 -10
View File
@@ -12,7 +12,7 @@ import (
type AccessTokenEntry struct {
Token string
DPoPKey *ecdsa.PrivateKey
Transport *DPoPTransport // Cache the transport to preserve nonce across requests
PDS string // Store PDS endpoint to create fresh transports
ExpiresAt time.Time
}
@@ -46,8 +46,10 @@ func (r *Refresher) GetAccessToken(ctx context.Context, did string) (string, *ec
r.mu.RUnlock()
if ok && time.Now().Before(entry.ExpiresAt) {
// Token still valid
return entry.Token, entry.DPoPKey, entry.Transport, nil
// Token still valid - create fresh transport to avoid nonce reuse
transport := NewDPoPTransport(nil, entry.DPoPKey)
transport.SetAccessToken(entry.Token)
return entry.Token, entry.DPoPKey, transport, nil
}
// Token expired or not cached, need to refresh
@@ -70,8 +72,10 @@ func (r *Refresher) GetAccessToken(ctx context.Context, did string) (string, *ec
r.mu.RUnlock()
if ok && time.Now().Before(entry.ExpiresAt) {
// Token was refreshed while we waited for the lock
return entry.Token, entry.DPoPKey, entry.Transport, nil
// Token was refreshed while we waited for the lock - create fresh transport
transport := NewDPoPTransport(nil, entry.DPoPKey)
transport.SetAccessToken(entry.Token)
return entry.Token, entry.DPoPKey, transport, nil
}
// Actually refresh the token
@@ -122,10 +126,7 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds
}
}
// Get DPoP transport (already has access token set by client.RefreshToken)
dpopTransport := client.DPoPTransport()
// Cache the access token and transport
// Cache the access token (but not transport - create fresh each time)
// Expire 1 minute early to avoid edge cases
expiresAt := token.Expiry.Add(-1 * time.Minute)
@@ -133,11 +134,15 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds
r.accessTokens[did] = &AccessTokenEntry{
Token: token.AccessToken,
DPoPKey: dpopKey,
Transport: dpopTransport, // Cache transport to preserve nonce
PDS: entry.PDS,
ExpiresAt: expiresAt,
}
r.mu.Unlock()
// Create fresh transport for this request
dpopTransport := NewDPoPTransport(nil, dpopKey)
dpopTransport.SetAccessToken(token.AccessToken)
return token.AccessToken, dpopKey, dpopTransport, nil
}
pkg/auth/oauth/server.go
+3 -3
View File
@@ -16,7 +16,7 @@ import (
// UISessionStore is the interface for UI session management
type UISessionStore interface {
Create(did, handle string, duration time.Duration) (string, error)
Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error)
}
// Server handles OAuth authorization for the AppView
@@ -172,8 +172,8 @@ func (s *Server) ServeCallback(w http.ResponseWriter, r *http.Request) {
// Check if this is a UI login (has oauth_return_to cookie)
if cookie, err := r.Cookie("oauth_return_to"); err == nil && s.uiSessionStore != nil {
// Create UI session
sessionID, err := s.uiSessionStore.Create(oauthState.DID, oauthState.Handle, 24*time.Hour)
// Create UI session with PDS endpoint
sessionID, err := s.uiSessionStore.Create(oauthState.DID, oauthState.Handle, oauthState.PDSEndpoint, 24*time.Hour)
if err != nil {
s.renderError(w, fmt.Sprintf("Failed to create UI session: %v", err))
return