debug: purge PDS

package router import ( "context" "database/sql" "encoding/json" "fmt" "log/slog" "net/http" "sync" "time" "github.com/bluesky-social/indigo/atproto/auth/oauth" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/delaneyj/toolbelt/embeddednats" "github.com/go-chi/chi/v5" "github.com/starfederation/datastar-go/datastar" "atmoquest/config" "atmoquest/features/about" "atmoquest/features/admin" "atmoquest/features/api" "atmoquest/features/auth" "atmoquest/features/connect" "atmoquest/features/connections" "atmoquest/features/demo" "atmoquest/features/event" "atmoquest/features/events" "atmoquest/features/index" "atmoquest/features/profile" "atmoquest/features/pwa" "atmoquest/features/settings" "atmoquest/features/tap" "atmoquest/internal/admincrypto" "atmoquest/internal/apitoken" "atmoquest/internal/connection" "atmoquest/internal/push" "atmoquest/internal/session" "atmoquest/web/resources" ) // SetupRoutes registers all top-level routes and feature subrouters. // // Shared dependencies are passed in explicitly rather than read from package // globals so tests can wire their own. As more features come online we may // promote this into a dedicated Deps struct. func SetupRoutes( ctx context.Context, router chi.Router, sess *session.Manager, oauthApp *oauth.ClientApp, ns *embeddednats.Server, conn *sql.DB, signer *admincrypto.Signer, ) (err error) { // Build the auth handlers once; other features depend on them for // RequireSession / ResumeSession. connQueue := connection.NewQueue(conn) authH := auth.NewHandlers(conn, oauthApp, sess) authH.ConnQueue = connQueue // Drain pending reciprocal connections on every authenticated page load, // debounced to once per 30 seconds per user. router.Use(auth.DrainMiddleware(authH, connQueue, conn, 30*time.Second)) // Real-time reciprocity: when a self-hosted Tap endpoint is configured, // subscribe to quest.atmo.connection creates and write/queue reciprocals // as they happen. No-op (and no goroutine) when TAP_WS_ENDPOINT is unset, // so dev/tests fall back to the login-drain path above. if config.Global.TapWSEndpoint != "" { tapConsumer := tap.NewConsumer( conn, oauthApp, connQueue, config.Global.TapWSEndpoint, config.Global.TapAuthToken, auth.AutoCheckinDrainHook(conn), slog.Default(), ) go tapConsumer.Run(ctx) } if config.Global.Environment == config.Dev { setupReload(router) setupDevDebug(router, authH, conn) } router.Handle("/static/*", resources.Handler()) auth.SetupRoutes(router, authH) profile.SetupRoutes(router, conn, authH) connect.SetupRoutes(router, conn, authH, connQueue) connections.SetupRoutes(router, conn, authH) event.SetupRoutes(router, conn, authH) events.SetupRoutes(router, conn, authH) admin.SetupRoutes(router, conn, authH, signer) settings.SetupRoutes(router, conn, authH) about.SetupRoutes(router, authH) pwa.SetupRoutes(router) demo.SetupRoutes(router) index.SetupRoutes(router, conn, authH) // JSON API for mobile app. tokens := apitoken.New(config.Global.SessionSecret) pusher := push.NewSender(conn) api.SetupRoutes(router, conn, authH, connQueue, tokens, pusher) router.Get("/healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ok")) }) // Recent badges endpoint for toast notifications. router.Get("/badges/recent", func(w http.ResponseWriter, r *http.Request) { didStr, _ := sess.Get(r) if didStr == "" { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"badges":[]}`)) return } did, err := syntax.ParseDID(didStr) if err != nil { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"badges":[]}`)) return } // Find badges earned in the last 60 seconds. rows, err := conn.QueryContext(r.Context(), ` SELECT d.name FROM earned_badges e JOIN badge_definitions d ON d.id = e.badge_id WHERE e.did = ? AND e.earned_at >= datetime('now', '-60 seconds') ORDER BY e.earned_at DESC `, did.String()) var names []string if err == nil { defer rows.Close() for rows.Next() { var name string if rows.Scan(&name) == nil { names = append(names, name) } } } w.Header().Set("Content-Type", "application/json") w.Header().Set("Cache-Control", "no-cache") _ = json.NewEncoder(w).Encode(map[string]any{"badges": names}) }) return nil } func setupReload(router chi.Router) { reloadChan := make(chan struct{}, 1) var hotReloadOnce sync.Once router.Get("/reload", func(w http.ResponseWriter, r *http.Request) { sse := datastar.NewSSE(w, r) reload := func() { sse.ExecuteScript("window.location.reload()") } hotReloadOnce.Do(reload) select { case <-reloadChan: reload() case <-r.Context().Done(): } }) router.Get("/hotreload", func(w http.ResponseWriter, r *http.Request) { select { case reloadChan <- struct{}{}: default: } w.WriteHeader(http.StatusOK) w.Write([]byte("OK")) }) } // setupDevDebug registers development-only debug routes. These are NEVER // available in production — the `config.Dev` gate in SetupRoutes ensures that. func setupDevDebug(router chi.Router, authH *auth.Handlers, db *sql.DB) { // POST /debug/purge-pds — delete all quest.atmo.* records from the // currently logged-in user's PDS and clear local cache tables. router.Post("/debug/purge-pds", func(w http.ResponseWriter, r *http.Request) { did, sess, ok := authH.RequireSession(w, r) if !ok { return } collections := []string{ "quest.atmo.connection", "quest.atmo.checkin", "quest.atmo.badge", "quest.atmo.event", "quest.atmo.profile", } type deleteResult struct { Collection string `json:"collection"` Deleted int `json:"deleted"` Errors int `json:"errors"` } var results []deleteResult apiClient := sess.APIClient() for _, coll := range collections { dr := deleteResult{Collection: coll} // List all records in this collection. cursor := "" for { params := map[string]any{ "repo": did.String(), "collection": coll, "limit": 100, } if cursor != "" { params["cursor"] = cursor } var resp struct { Records []struct { URI string `json:"uri"` } `json:"records"` Cursor string `json:"cursor,omitempty"` } if err := apiClient.Get(r.Context(), syntax.NSID("com.atproto.repo.listRecords"), params, &resp); err != nil { slog.Warn("debug purge: list", "collection", coll, "err", err) break } for _, rec := range resp.Records { rkey := rkeyFromURI(rec.URI) if rkey == "" { dr.Errors++ continue } input := map[string]any{ "repo": did.String(), "collection": coll, "rkey": rkey, } if err := apiClient.Post(r.Context(), syntax.NSID("com.atproto.repo.deleteRecord"), input, nil); err != nil { slog.Warn("debug purge: delete", "uri", rec.URI, "err", err) dr.Errors++ } else { dr.Deleted++ } } if resp.Cursor == "" || len(resp.Records) == 0 { break } cursor = resp.Cursor } results = append(results, dr) } // Also clear local cache tables. for _, stmt := range []string{ `DELETE FROM checkins WHERE did = ?`, `DELETE FROM pending_connections WHERE target_did = ? OR initiator_did = ?`, `DELETE FROM connection_notes WHERE viewer_did = ?`, } { if stmt == `DELETE FROM pending_connections WHERE target_did = ? OR initiator_did = ?` { _, _ = db.ExecContext(r.Context(), stmt, did.String(), did.String()) } else { _, _ = db.ExecContext(r.Context(), stmt, did.String()) } } slog.Info("debug purge complete", "did", did.String(), "results", fmt.Sprintf("%+v", results)) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "did": did.String(), "results": results, }) }) // GET /debug/purge-pds — simple confirmation page. router.Get("/debug/purge-pds", func(w http.ResponseWriter, r *http.Request) { didStr, _ := authH.Sessions.Get(r) w.Header().Set("Content-Type", "text/html; charset=utf-8") fmt.Fprintf(w, ` debug: purge PDS

⚠ Purge all quest.atmo.* records

DID: %s

This will delete all quest.atmo.connection, checkin, badge, event, and profile records from your PDS, plus clear local cache tables.

`, didStr) }) } // rkeyFromURI extracts the rkey (last path segment) from an at-uri like // "at://did:plc:xxx/quest.atmo.connection/3abc123". func rkeyFromURI(uri string) string { // at-uri format: at://// for i := len(uri) - 1; i >= 0; i-- { if uri[i] == '/' { return uri[i+1:] } } return "" }