diff --git a/go/cmd/emailnotifierjob/main.go b/go/cmd/emailnotifierjob/main.go index b16ba04..14be3e6 100644 --- a/go/cmd/emailnotifierjob/main.go +++ b/go/cmd/emailnotifierjob/main.go @@ -20,22 +20,17 @@ import ( ) const ( - // Only notify about streams with activity in the last 24 hours - maxActivityAge = 24 * time.Hour - // Minimum time a message must be unread before we consider notifying - unreadThreshold = 10 * time.Minute - // Minimum time between emails to the same user - emailCooldown = 12 * time.Hour + maxActivityAge = 24 * time.Hour // ignore streams idle longer than this + unreadThreshold = 10 * time.Minute // grace window before a message is "unread" + emailCooldown = 12 * time.Hour // min gap between emails to the same user ) func main() { ctx := context.Background() - // Initialize Postgres db.Init() defer db.Cleanup() - // Initialize Firestore gcpProject := utils.MustGetEnv("GCP_PROJECT") firestoreClient, err := firestore.NewClient(ctx, gcpProject) if err != nil { @@ -44,7 +39,6 @@ func main() { } defer firestoreClient.Close() - // Initialize aero (email) gRPC client aeroAddr := utils.MustGetEnv("AERO_ADDR") aeroConn, err := grpc.NewClient(aeroAddr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { @@ -54,7 +48,6 @@ func main() { defer aeroConn.Close() aeroSvc := pbaero.NewPrimaryClient(aeroConn) - // Initialize pusher gRPC client pusherAddr := utils.MustGetEnv("PUSHER_GRPC_ADDR") pusherConn, err := grpc.NewClient(pusherAddr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { @@ -64,7 +57,6 @@ func main() { defer pusherConn.Close() pusherSvc := pbpusher.NewPusherServiceClient(pusherConn) - // Initialize services humanSvc := human.NewService(db.Pool()) networkSvc := network.NewReader(db.Pool()) @@ -86,13 +78,11 @@ func runNotificationCycle( ) error { now := time.Now() - // Load all networks networks, err := networkReader.ListAll(ctx) if err != nil { return fmt.Errorf("listing networks: %w", err) } - // Load all humans into a lookup map allHumans, err := humanSvc.ListAll(ctx) if err != nil { return fmt.Errorf("listing humans: %w", err) @@ -102,11 +92,9 @@ func runNotificationCycle( humansById[h.ID] = h } - // Track which streams each human is behind on, and the latest activity across those streams behindCounts := map[string]int{} latestActivity := map[string]time.Time{} - // Get all currently connected humans allOnline := map[string]bool{} onlineResp, err := pusherSvc.GetOnlineHumanIds(ctx, &pbpusher.GetOnlineHumanIdsRequest{}) if err != nil { @@ -118,35 +106,29 @@ func runNotificationCycle( } for _, net := range networks { - // Query Firestore for open streams in this network streams, err := getOpenStreams(ctx, fsClient, net.ID) if err != nil { slog.Error("failed to query streams", "networkId", net.ID, "error", err) continue } - // net.MemberHumanIds already contains the admin (Create() adds them and - // RemoveMemberFromNetwork won't drop them) — no need to union separately. + // net.MemberHumanIds already includes the admin. for _, stream := range streams { if stream.LastChildCreatedAt == nil { continue } - // Skip streams with no recent activity if now.Sub(*stream.LastChildCreatedAt) > maxActivityAge { continue } - // Skip if the latest message is too fresh (within threshold) if now.Sub(*stream.LastChildCreatedAt) < unreadThreshold { continue } - // Resolve members from visible_to using the shared helper for _, humanId := range network.ResolveVisibility(stream.VisibleTo, net.MemberHumanIds) { marker, hasMarker := stream.PlaybackMarkers[humanId] if hasMarker && !marker.Before(*stream.LastChildCreatedAt) { - continue // up to date + continue } - // No marker or marker is behind → this human is behind on this stream behindCounts[humanId]++ if stream.LastChildCreatedAt.After(latestActivity[humanId]) { latestActivity[humanId] = *stream.LastChildCreatedAt @@ -156,10 +138,8 @@ func runNotificationCycle( } - // Send notifications sentCount := 0 for humanId, count := range behindCounts { - // Skip online users if allOnline[humanId] { slog.Info("human online...skipping email", "humanId", humanId) continue @@ -170,28 +150,24 @@ func runNotificationCycle( continue } - // Skip if notifications disabled if !h.EmailNotificationsEnabled { continue } - // Skip if no new activity since last notification + // Skip if nothing new since the previous email. if h.LastEmailNotificationSentAt != nil && !latestActivity[humanId].After(*h.LastEmailNotificationSentAt) { continue } - // Enforce cooldown between emails to the same user if h.LastEmailNotificationSentAt != nil && now.Sub(*h.LastEmailNotificationSentAt) < emailCooldown { continue } - // Send email if err := sendNotificationEmail(ctx, aeroSvc, h, count); err != nil { slog.Error("failed to send email", "humanId", humanId, "error", err) continue } - // Update last sent timestamp if err := humanSvc.UpdateLastEmailNotificationSentAt(ctx, humanId, now); err != nil { slog.Error("failed to update last_email_notification_sent_at", "humanId", humanId, "error", err) } @@ -207,7 +183,6 @@ func runNotificationCycle( return nil } -// getOpenStreams queries Firestore for all open stream particles in a network. func getOpenStreams(ctx context.Context, client *firestore.Client, networkId string) ([]particle.FirestoreStreamParticle, error) { collPath := fmt.Sprintf("networks/%s/children", networkId) docs, err := client.Collection(collPath). diff --git a/go/cmd/memberreconciler/main.go b/go/cmd/memberreconciler/main.go index a1d1c7c..02b4528 100644 --- a/go/cmd/memberreconciler/main.go +++ b/go/cmd/memberreconciler/main.go @@ -1,13 +1,7 @@ -// memberreconciler is a one-shot job (also safe to run on a cron) that makes -// the Firestore membership mirror (humans/{humanId}.networks) match the -// authoritative Postgres network_members table. -// -// Run on a cron to heal any drift from a dropped mirror -// write in network.Service. -// -// The reconciler reads the current Firestore state and only writes humans -// whose mirrored networks differ from Postgres. Writes cost ~3x reads, and in -// steady state drift is rare, so read-first keeps the cron nearly free. +// memberreconciler reconciles the Firestore membership mirror +// (humans/{humanId}.networks) against the authoritative Postgres +// network_members table. Safe to run on a cron — only humans whose mirrored +// set differs from Postgres are written, so a steady-state run is nearly free. package main import ( @@ -98,8 +92,7 @@ func reconcile( return written, scanned, nil } -// snapshotMirror streams the humans collection once and returns a map of -// humanId -> current networks array. One iterator, N billed reads. +// One iterator, N billed reads — returns humanId → mirrored networks. func snapshotMirror(ctx context.Context, fs *firestore.Client) (map[string][]string, error) { out := map[string][]string{} iter := fs.Collection("humans").Documents(ctx) @@ -124,9 +117,8 @@ func snapshotMirror(ctx context.Context, fs *firestore.Client) (map[string][]str return out, nil } -// sameSet reports whether a and b contain the same elements, ignoring order -// and duplicates. Firestore array ops don't preserve order, so set equality is -// the right comparison for the networks array. +// Set equality (order- and duplicate-insensitive); Firestore array ops don't +// preserve order. func sameSet(a, b []string) bool { if len(a) == 0 && len(b) == 0 { return true diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index 0f6c4eb..d6a5be6 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -185,8 +185,7 @@ func main() { mux.Handle("GET /waitlist/{email}", withAuth(h.GetWaitlistEntry)) mux.Handle("POST /waitlist/invite", withAuth(h.InviteWaitlistEntrant)) - // Apply middleware - // nil allows all origins (required for electron app) + // nil = allow all origins (Electron app needs it). muxWithCors := middleware.CORS(nil)(mux) addr := fmt.Sprintf("0.0.0.0:%s", port) diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index b18706d..a9a3e97 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -38,13 +38,9 @@ func createClient(ctx context.Context) *firestore.Client { return client } -// The purpose of the particle processor worker is to listen for new particles -// across all streams and perform side effects such as -// - generate transcript if the particle is of type media -// - send mobile notifications if a client is offline -// - update the parent stream's `last_child_created_at` -// - generate vector embedding -// - synthesize and decide whether ai should generate a particle as a response +// Listens for new particles and runs per-particle side effects: transcripts, +// transcode, parent stream's last_child_created_at, freemium usage, and push +// notifications for offline recipients. func main() { ctx := context.Background() @@ -80,9 +76,8 @@ func main() { humanSvc := human.NewService(db.Pool()) networkReader := network.NewReader(db.Pool()) pushTokenSvc := pushnotify.NewService(db.Pool()) - // EXPO_ACCESS_TOKEN is required: with Enhanced Security enabled on the Expo - // project, sends without it fail; without it, anyone holding one of our - // push tokens could spam our users via the public Expo endpoint. + // EXPO_ACCESS_TOKEN is required: Enhanced Security is on for our Expo + // project (otherwise anyone holding one of our push tokens could spam users). expoClient := pushnotify.NewExpoClient(utils.MustGetEnv("EXPO_ACCESS_TOKEN")) notifier := pushnotify.NewNotifier(networkReader, pushTokenSvc, pusherClient, expoClient) @@ -127,8 +122,8 @@ func main() { slog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data()) - // --- Perform side effects --- - // All of them do not stop us from marking the particle as processed. + // Side effects below are best-effort — failures don't prevent + // marking the particle as processed. parentDoc := loadParentParticle(ctx, change.Doc) updateParentLastChildCreatedAt(ctx, change.Doc, parentDoc) transcript := transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) @@ -143,9 +138,8 @@ func main() { } } -// transcribeMediaParticle transcribes a media particle, writes the structured -// transcript to Firestore, and returns the raw transcript text. Returns "" for -// non-media particles or on any error (errors are logged internally). +// Writes the structured transcript to Firestore and returns the raw text; +// returns "" for non-media particles or on any error (logged internally). func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speechSvc speech.SpeechService, doc *firestore.DocumentSnapshot) string { var mediaParticle particle.FirestoreMediaParticle err := doc.DataTo(&mediaParticle) @@ -227,10 +221,9 @@ func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTr } } -// recordFreemiumUsage bumps the network's daily message counter for non-container -// particles. Idempotent via the surrounding processed_particles guard: the worker -// only reaches this path on first-seen particles, so a crash/restart won't -// double-count. +// Bumps the network's daily message counter for non-container particles. +// The surrounding processed_particles guard keeps this idempotent across +// crashes/restarts. func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *firestore.DocumentSnapshot) { rawType, err := doc.DataAt("type") if err != nil { @@ -247,7 +240,7 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f slog.Error("invalid particle type", "error", err, "particleID", doc.Ref.ID) return } - // Containers (stream/folder) don't count as "messages" for the daily cap. + // Containers don't count toward the daily message cap. if particleType == particle.TypeStream || particleType == particle.TypeFolder { return } @@ -263,8 +256,7 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f } } -// loadParentParticle fetches the immediate parent particle doc for `doc`. -// Returns nil (and logs) if the path doesn't have a parent or the read fails. +// Returns nil (and logs) if the path has no parent or the read fails. func loadParentParticle(ctx context.Context, doc *firestore.DocumentSnapshot) *firestore.DocumentSnapshot { parentChildrenCollectionRef := doc.Ref.Parent if parentChildrenCollectionRef == nil { @@ -283,9 +275,8 @@ func loadParentParticle(ctx context.Context, doc *firestore.DocumentSnapshot) *f return parentParticleDoc } -// updateParentLastChildCreatedAt updates the parent stream's last_child_created_at -// to the child's actual created_at timestamp, so it stays directly comparable with -// playback markers (which also store child created_at values). +// Sets last_child_created_at to the child's created_at so it stays directly +// comparable with playback markers (which also store child created_at values). func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.DocumentSnapshot, parent *firestore.DocumentSnapshot) { if parent == nil { return @@ -307,7 +298,6 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document return } - // Read the child's created_at — this is the same value that playback markers store childCreatedAt, err := doc.DataAt("created_at") if err != nil { slog.Error("failed to read child created_at", "error", err, "particleID", doc.Ref.ID) @@ -325,10 +315,9 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document } } -// notifyForParticle dispatches a push notification for a newly-created particle. -// Skips containers (streams/folders) and particles whose parent isn't a stream -// (notifications are only sent for stream messages today). The transcript arg -// is used as the preview body for media particles when available. +// Skips containers and particles whose parent isn't a stream — notifications +// are scoped to stream messages today. The transcript arg becomes the preview +// body for media particles when available. func notifyForParticle( ctx context.Context, notifier *pushnotify.Notifier, @@ -405,11 +394,8 @@ func notifyForParticle( } } -// previewForParticle builds the visible notification body. Kept short — push -// previews truncate aggressively on lockscreens. For media, prefers the -// transcript text (already computed by transcribeMediaParticle in the same -// processing step) and falls back to the generic "Sent a ..." line if speech -// recognition produced nothing. +// Builds the notification body. Kept short — lockscreens truncate aggressively. +// Media prefers transcript text and falls back to a generic "Sent a …" line. func previewForParticle(pType particle.ParticleType, doc *firestore.DocumentSnapshot, transcript string) string { switch pType { case particle.TypeText: diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 53a0ae3..2dee5ca 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -25,29 +25,22 @@ func main() { port := utils.MustGetEnv("PORT") grpcPort := utils.MustGetEnv("GRPC_PORT") - // Initialize database (for network membership checks) db.Init() defer db.Cleanup() - // Redis for auth session validation (same DB as orion) - authRedis := internal.ConnectAndTestRedis(db.RedisDBAuth) + authRedis := internal.ConnectAndTestRedis(db.RedisDBAuth) // shared with orion + pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) // presence + pub/sub - // Redis for pusher state (presence hashes, pub/sub) - pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) + sessionReader := auth.NewSessionReader(authRedis) + networkReader := network.NewReader(db.Pool()) - // Services - sessionReader := auth.NewSessionReader(authRedis) // pusher only validates sessions - networkReader := network.NewReader(db.Pool()) // pusher only checks membership - - // Pod identity (use hostname in k8s, which is the pod name) + // Hostname is the k8s pod name. podID, err := os.Hostname() if err != nil { podID = fmt.Sprintf("pod-%d", os.Getpid()) } - // Pusher core bridge := pusher.NewRedisBridge(pusherRedis, podID) - // Context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -56,16 +49,11 @@ func main() { bridge.SetHub(hub) server := pusher.NewServer(ctx, hub, bridge, sessionReader) - // Start hub event loop go hub.Run(ctx) - - // Start Redis Pub/Sub listener go bridge.Listen(ctx) - - // Start pod heartbeat + stale pod cleanup go bridge.Heartbeat(ctx) - // --- gRPC server (internal, for presence queries) --- + // --- gRPC server (internal presence queries) --- grpcListener, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%s", grpcPort)) if err != nil { slog.Error("failed to listen for gRPC", "port", grpcPort, "error", err) @@ -104,7 +92,7 @@ func main() { <-sigCh slog.Info("shutting down...") - cancel() // stops hub, bridge listener, heartbeat + cancel() grpcServer.GracefulStop() httpServer.Shutdown(context.Background()) diff --git a/go/cmd/transcodebackfill/main.go b/go/cmd/transcodebackfill/main.go index 258069e..df931df 100644 --- a/go/cmd/transcodebackfill/main.go +++ b/go/cmd/transcodebackfill/main.go @@ -1,12 +1,6 @@ -// transcodebackfill is a one-shot job that scans every doc under the -// "children" Firestore collection group, identifies media particles missing a -// transcoded variant, and re-runs the transcode + upload + Firestore-update -// flow against the configured GCS bucket. Idempotent — re-running the Job -// after a partial failure picks up where it left off via the existing -// `transcoded_object_id != ""` short-circuit inside particle.Transcode. -// -// Intended to back-fill the production iOS playback backlog created before we -// launched mobile / particle processor worker only transcodes new media. +// transcodebackfill walks every doc under the "children" collection group and +// re-runs transcode for media particles missing a transcoded variant. +// Idempotent: particle.Transcode short-circuits on transcoded_object_id != "". package main import ( @@ -104,9 +98,7 @@ func run(ctx context.Context, fs *firestore.Client, depotSvc depot.Service) (sta s.scanned++ - // Cheap pre-filter: most docs under the "children" collection group - // are not media particles. DataAt avoids unmarshalling the full - // document for those. + // DataAt avoids unmarshalling the full doc; most children aren't media. rawType, err := doc.DataAt("type") if err != nil { s.skippedNonMedia++ diff --git a/go/internal/auth/reader.go b/go/internal/auth/reader.go index 037b637..0a5a716 100644 --- a/go/internal/auth/reader.go +++ b/go/internal/auth/reader.go @@ -18,9 +18,8 @@ type sessionReaderImpl struct { redisClient *redis.Client } -// newSessionReader returns the concrete reader. Used by NewAuthService to -// embed without going through the SessionReader interface (which would hide -// redisClient from the rest of authServiceImpl). +// Exposes the concrete type so authServiceImpl can embed it without +// hiding redisClient behind the SessionReader interface. func newSessionReader(redisClient *redis.Client) *sessionReaderImpl { return &sessionReaderImpl{redisClient: redisClient} } diff --git a/go/internal/auth/service.go b/go/internal/auth/service.go index 40b12a9..244e35c 100644 --- a/go/internal/auth/service.go +++ b/go/internal/auth/service.go @@ -47,17 +47,16 @@ type Session struct { type AuthService interface { SessionReader - // RequestSignInCode generates a code and emails it to the provided email. - // To retrieve a session, client must verify with VerifySignInCode. + // RequestSignInCode emails a one-time code; the client redeems it via VerifySignInCode. RequestSignInCode(ctx context.Context, email string) error - // VerifySignInCode returns ErrInvalidCode if incorrect code, otherwise creates a session. - // humanId is stored in the session alongside the email. + // VerifySignInCode returns ErrInvalidCode on a wrong code, otherwise creates + // a session keyed to (email, humanId). VerifySignInCode(ctx context.Context, email, code, humanId string) (sessionToken string, err error) - // ExtendSession returns ErrSessionNotFound if no valid session + // ExtendSession returns ErrSessionNotFound if no valid session. ExtendSession(ctx context.Context, sessionToken string) error SignOut(ctx context.Context, sessionToken string) error - // MintFirebaseCustomToken returns a Firebase custom token with uid=humanId and no custom claims. + // MintFirebaseCustomToken issues a Firebase custom token with uid=humanId and no claims. MintFirebaseCustomToken(ctx context.Context, humanId string) (string, error) IsSystemAdmin(ctx context.Context, email string) bool diff --git a/go/internal/billing/service.go b/go/internal/billing/service.go index fabea1a..9bb1164 100644 --- a/go/internal/billing/service.go +++ b/go/internal/billing/service.go @@ -73,9 +73,8 @@ func NewService(ctx context.Context, pool *pgxpool.Pool, cfg Config) (Service, e }, nil } -// NewServiceForWorker builds a minimal billing Service suitable for the -// particle processor worker: only the usage-tracking path is exercised, so -// we skip Stripe client setup (no API key required). +// NewServiceForWorker skips Stripe client setup since workers only exercise +// the usage-tracking path. No API key required. func NewServiceForWorker(pool *pgxpool.Pool) Service { return &serviceImpl{ usageRepo: newUsageRepository(pool), diff --git a/go/internal/billing/usage_models.go b/go/internal/billing/usage_models.go index 6e8ec00..0787edc 100644 --- a/go/internal/billing/usage_models.go +++ b/go/internal/billing/usage_models.go @@ -2,11 +2,9 @@ package billing import "time" -// FreemiumDailyLimit is the per-network daily cap on usage, -// agnostic of the units that this refer to. This is only relevant for the "free" plan. +// Per-network daily cap on the free plan. Unit-agnostic. const FreemiumDailyLimit = 50 -// Usage describes a network's current freemium quota state for today. type Usage struct { Plan Plan `json:"plan"` Used int `json:"used"` diff --git a/go/internal/db/redis.go b/go/internal/db/redis.go index d663685..84a847e 100644 --- a/go/internal/db/redis.go +++ b/go/internal/db/redis.go @@ -1,8 +1,8 @@ package db -// Shared database namespaces used across services -// FIX: Use separate redis instance. We start with higher number because use this same instance in helios. +// FIX: move to a dedicated Redis instance. The high DB numbers exist because +// this instance is shared with helios. const ( RedisDBAuth = 4 // auth sessions - RedisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) + RedisDBPusher = 5 // presence, pub/sub ) diff --git a/go/internal/depot/models.go b/go/internal/depot/models.go index 8042a72..219e921 100644 --- a/go/internal/depot/models.go +++ b/go/internal/depot/models.go @@ -2,7 +2,6 @@ package depot import "time" -// Object represents a stored object in the depot type Object struct { ID string Name string @@ -14,31 +13,26 @@ type Object struct { CreatedAt time.Time } -// PrepareUploadInput represents the input for preparing an upload type PrepareUploadInput struct { - Prefix string // Optional prefix for organizing objects (e.g., network_id) + Prefix string // optional, e.g. network_id Name string ContentType string ContentLength int64 } -// PrepareUploadResult represents the result of preparing an upload type PrepareUploadResult struct { ObjectID string UploadURL string UploadHeaders map[string]string } -// CreateFromReaderInput is for server-side direct uploads (no presigned URL). -// Used by background workers that already have the bytes on hand and don't -// need a client round-trip. +// For server-side direct uploads (no presigned URL). type CreateFromReaderInput struct { - Prefix string // Optional prefix for organizing objects (e.g., network_id) + Prefix string // optional, e.g. network_id Name string ContentType string } -// Config holds configuration for the depot service type Config struct { GoogleServiceAccountEmail string BucketName string diff --git a/go/internal/depot/service.go b/go/internal/depot/service.go index 4026552..c87221d 100644 --- a/go/internal/depot/service.go +++ b/go/internal/depot/service.go @@ -74,10 +74,10 @@ func (s *serviceImpl) PrepareUpload(ctx context.Context, input PrepareUploadInpu return nil, errors.Join(ErrInvalidInput, errors.New("content_length must be positive")) } - // Generate object key: {prefix}/{uuid}/{filename} + // {prefix}/{uuid}/{filename} objectKey := fmt.Sprintf("%s/%s/%s", input.Prefix, uuid.New().String(), input.Name) - // Create the database record (contains_content = false initially) + // Row is written first with contains_content=false; ConfirmUpload flips it. obj := &Object{ Name: input.Name, ContentType: input.ContentType, @@ -92,8 +92,7 @@ func (s *serviceImpl) PrepareUpload(ctx context.Context, input PrepareUploadInpu return nil, err } - // Generate a signed URL for uploading with Content-Length enforcement - // The Headers field specifies headers that MUST be included in the upload request + // Content-Length is part of the signature, so the client must send it verbatim. contentLengthHeader := fmt.Sprintf("Content-Length:%d", input.ContentLength) uploadURL, err := s.storageClient.Bucket(s.bucketName).SignedURL(objectKey, &storage.SignedURLOptions{ GoogleAccessID: s.googleServiceAccountEmail, @@ -104,7 +103,7 @@ func (s *serviceImpl) PrepareUpload(ctx context.Context, input PrepareUploadInpu }) if err != nil { slog.Error("failed to generate signed upload URL", "error", err, "bucket", s.bucketName, "object_key", objectKey) - // Clean up the database record if we can't generate the URL + // Roll back the placeholder row. if delErr := s.repo.delete(ctx, created.ID); delErr != nil { slog.Warn("failed to cleanup db record after signed URL failure", "error", delErr, "object_id", created.ID) } @@ -130,7 +129,6 @@ func (s *serviceImpl) ConfirmUpload(ctx context.Context, objectID string) (*Obje return nil, err } - // Verify the object exists in GCS and check its size matches expected attrs, err := s.storageClient.Bucket(obj.BucketName).Object(obj.ObjectKey).Attrs(ctx) if err != nil { if errors.Is(err, storage.ErrObjectNotExist) { @@ -140,12 +138,10 @@ func (s *serviceImpl) ConfirmUpload(ctx context.Context, objectID string) (*Obje return nil, err } - // Verify content length matches what was declared if attrs.Size != obj.ContentLength { return nil, errors.Join(ErrInvalidInput, fmt.Errorf("content length mismatch: expected %d, got %d", obj.ContentLength, attrs.Size)) } - // Mark as containing content if err := s.repo.setContainsContent(ctx, objectID, true); err != nil { if errors.Is(err, errNotFound) { return nil, ErrNotFound @@ -153,14 +149,12 @@ func (s *serviceImpl) ConfirmUpload(ctx context.Context, objectID string) (*Obje return nil, err } - // Fetch and return the updated object return s.repo.getByID(ctx, objectID) } -// CreateFromReader streams bytes directly to GCS using the storage client and -// records the depot_objects row in one shot. Unlike PrepareUpload, there is no -// signed URL or client round-trip — the caller already has the bytes. Intended -// for worker-side flows (e.g. transcoded media variants). +// CreateFromReader streams bytes straight to GCS and writes the row in one +// shot — no signed URL, no client round-trip. For server-side flows that +// already have the bytes (e.g. transcoded variants). func (s *serviceImpl) CreateFromReader(ctx context.Context, input CreateFromReaderInput, body io.Reader) (*Object, error) { if input.Name == "" { return nil, errors.Join(ErrInvalidInput, errors.New("name is required")) @@ -174,7 +168,7 @@ func (s *serviceImpl) CreateFromReader(ctx context.Context, input CreateFromRead w := s.storageClient.Bucket(s.bucketName).Object(objectKey).NewWriter(ctx) w.ContentType = input.ContentType if _, err := io.Copy(w, body); err != nil { - // Close to release resources, then surface the original copy error. + // Always release the writer; surface the copy error, not Close's. if cerr := w.Close(); cerr != nil { slog.Warn("failed to close GCS writer after copy failure", "error", cerr, "object_key", objectKey) } @@ -197,7 +191,7 @@ func (s *serviceImpl) CreateFromReader(ctx context.Context, input CreateFromRead created, err := s.repo.create(ctx, obj) if err != nil { - // Best-effort: clean up the GCS object since we can't track it in the DB. + // Best-effort: drop the now-untracked GCS object. if delErr := s.storageClient.Bucket(s.bucketName).Object(objectKey).Delete(ctx); delErr != nil { slog.Warn("failed to clean up GCS object after db create failure", "error", delErr, "object_key", objectKey) } @@ -227,7 +221,6 @@ func (s *serviceImpl) GetDownloadURL(ctx context.Context, objectID string) (stri return "", err } - // Generate a signed URL for downloading downloadURL, err := s.storageClient.Bucket(obj.BucketName).SignedURL(obj.ObjectKey, &storage.SignedURLOptions{ GoogleAccessID: s.googleServiceAccountEmail, Method: "GET", @@ -250,14 +243,13 @@ func (s *serviceImpl) Delete(ctx context.Context, objectID string) error { return err } - // Delete from GCS (ignore not found errors) + // GCS first so we don't strand an object after the row vanishes; missing object is fine. gcsErr := s.storageClient.Bucket(obj.BucketName).Object(obj.ObjectKey).Delete(ctx) if gcsErr != nil && !errors.Is(gcsErr, storage.ErrObjectNotExist) { slog.Error("failed to delete object from GCS", "error", gcsErr, "bucket", obj.BucketName, "object_key", obj.ObjectKey) return gcsErr } - // Delete from database if err := s.repo.delete(ctx, objectID); err != nil { if errors.Is(err, errNotFound) { return ErrNotFound diff --git a/go/internal/handler/billing.go b/go/internal/handler/billing.go index 2d17206..23423ec 100644 --- a/go/internal/handler/billing.go +++ b/go/internal/handler/billing.go @@ -25,12 +25,8 @@ type PortalSessionResponse struct { URL string `json:"url"` } -// GetNetworkUsage returns the freemium quota state for the authenticated -// caller's current network: how many messages they've used today, the daily -// limit (null for pro), and when the counter resets. -// -// Authorization: any network member may read (not admin-only) since the UI -// surfaces this to every sender. +// GetNetworkUsage reports today's usage, daily limit (nil on pro), and reset +// time. Open to any network member since the UI surfaces it to every sender. func (h *Handler) GetNetworkUsage(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -170,8 +166,8 @@ func (h *Handler) HandleStripeWebhook(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// loadNetworkForAdmin resolves the {id} path param and verifies the caller -// is the network's admin. On failure it writes the HTTP error and returns ok=false. +// Resolves {id}, verifies the caller is admin. On failure writes the HTTP +// error and returns ok=false. func (h *Handler) loadNetworkForAdmin(w http.ResponseWriter, r *http.Request) (*network.Network, string, bool) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { diff --git a/go/internal/handler/handler.go b/go/internal/handler/handler.go index dd565dd..d87a62c 100644 --- a/go/internal/handler/handler.go +++ b/go/internal/handler/handler.go @@ -172,7 +172,7 @@ type DepotObject struct { // Auth Handlers // ============================================================================ -// RequestSignInCode creates a human account if not already existent and sends a sign-in code +// RequestSignInCode auto-creates the human if missing, then emails a one-time code. func (h *Handler) RequestSignInCode(w http.ResponseWriter, r *http.Request) { var req RequestSignInCodeRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -185,7 +185,6 @@ func (h *Handler) RequestSignInCode(w http.ResponseWriter, r *http.Request) { return } - // Auto-create human if doesn't exist _, err := h.humanSvc.GetOrCreateByEmail(r.Context(), req.Email) if err != nil { slog.Error("failed to get or create human", "error", err, "email", req.Email) @@ -193,7 +192,6 @@ func (h *Handler) RequestSignInCode(w http.ResponseWriter, r *http.Request) { return } - // Request sign-in code if err := h.authSvc.RequestSignInCode(r.Context(), req.Email); err != nil { slog.Error("failed to request sign-in code", "error", err, "email", req.Email) http.Error(w, "internal server error", http.StatusInternalServerError) @@ -203,7 +201,6 @@ func (h *Handler) RequestSignInCode(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } -// SignIn verifies the code and returns a session token func (h *Handler) SignIn(w http.ResponseWriter, r *http.Request) { var req SignInRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -216,7 +213,7 @@ func (h *Handler) SignIn(w http.ResponseWriter, r *http.Request) { return } - // Look up human first so we can store humanId in the session + // humanId is captured into the session so later requests don't re-resolve email → id. hum, err := h.humanSvc.GetByEmail(r.Context(), req.Email) if err != nil { if errors.Is(err, human.ErrNotFound) { @@ -248,9 +245,8 @@ func (h *Handler) SignIn(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// FirebaseToken mints a Firebase custom token for the authenticated human so -// the client can signInWithCustomToken and have request.auth.uid populated in -// Firestore security rules. +// FirebaseToken mints a custom token so the client can signInWithCustomToken +// and have request.auth.uid populated in Firestore security rules. func (h *Handler) FirebaseToken(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -269,7 +265,6 @@ func (h *Handler) FirebaseToken(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(FirebaseTokenResponse{Token: token}) } -// SignOut deletes the session from the token in headers func (h *Handler) SignOut(w http.ResponseWriter, r *http.Request) { token := extractBearerToken(r) if token == "" { @@ -286,7 +281,6 @@ func (h *Handler) SignOut(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } -// GetCurrentHuman returns the authenticated human func (h *Handler) GetCurrentHuman(w http.ResponseWriter, r *http.Request) { email, ok := middleware.EmailFromContext(r.Context()) if !ok { @@ -314,7 +308,6 @@ type UpdateSettingsRequest struct { EmailNotificationsEnabled *bool `json:"email_notifications_enabled"` } -// UpdateSettings updates the authenticated human's settings func (h *Handler) UpdateSettings(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -343,7 +336,6 @@ func (h *Handler) UpdateSettings(w http.ResponseWriter, r *http.Request) { // Network Handlers // ============================================================================ -// CreateNetwork creates a new network func (h *Handler) CreateNetwork(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -380,7 +372,6 @@ func (h *Handler) CreateNetwork(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// ListNetworks retrieves networks for the authenticated human func (h *Handler) ListNetworks(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -409,7 +400,6 @@ func (h *Handler) ListNetworks(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// GetNetwork retrieves a specific network func (h *Handler) GetNetwork(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -456,8 +446,8 @@ func (h *Handler) GetNetwork(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// AddMembersToNetwork adds members to a network. Registered users are added as members, -// unregistered users receive email invitations. +// AddMembersToNetwork routes registered users into membership and emails an +// invitation to the rest. func (h *Handler) AddMembersToNetwork(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -493,7 +483,6 @@ func (h *Handler) AddMembersToNetwork(w http.ResponseWriter, r *http.Request) { return } - // Resolve emails: registered users become members, unregistered get invitations var memberHumanIds []string var inviteEmails []string for _, email := range req.EmailAddresses { @@ -531,7 +520,6 @@ func (h *Handler) AddMembersToNetwork(w http.ResponseWriter, r *http.Request) { } } - // Return updated network net, err := h.networkSvc.GetByID(r.Context(), networkID) if err != nil { slog.Error("failed to get network after adding members", "error", err, "network_id", networkID) @@ -550,9 +538,8 @@ func (h *Handler) AddMembersToNetwork(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// RemoveMemberFromNetwork removes a member from a network. Admin-only. -// Admins cannot remove themselves — doing so would leave networks.admin_human_id -// dangling. Removal of a non-member is a no-op (204). +// RemoveMemberFromNetwork is admin-only. Admins cannot remove themselves +// (would orphan networks.admin_human_id); removing a non-member is a no-op (204). func (h *Handler) RemoveMemberFromNetwork(w http.ResponseWriter, r *http.Request) { net, _, ok := h.loadNetworkForAdmin(w, r) if !ok { @@ -579,7 +566,6 @@ func (h *Handler) RemoveMemberFromNetwork(w http.ResponseWriter, r *http.Request w.WriteHeader(http.StatusNoContent) } -// ListInvitationsForNetwork returns pending invitations for a network func (h *Handler) ListInvitationsForNetwork(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -625,7 +611,6 @@ func (h *Handler) ListInvitationsForNetwork(w http.ResponseWriter, r *http.Reque json.NewEncoder(w).Encode(resp) } -// ListMyInvitations returns pending invitations for the authenticated user func (h *Handler) ListMyInvitations(w http.ResponseWriter, r *http.Request) { email, ok := middleware.EmailFromContext(r.Context()) if !ok { @@ -654,7 +639,6 @@ func (h *Handler) ListMyInvitations(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// AcceptInvitation accepts a pending network invitation for the authenticated user func (h *Handler) AcceptInvitation(w http.ResponseWriter, r *http.Request) { email, ok := middleware.EmailFromContext(r.Context()) if !ok { @@ -687,7 +671,6 @@ func (h *Handler) AcceptInvitation(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } -// RevokeInvitation revokes a pending invitation from a network func (h *Handler) RevokeInvitation(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -732,7 +715,7 @@ func (h *Handler) RevokeInvitation(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } -// DownloadParticleMedia returns a fresh signed download URL for media/file particles +// DownloadParticleMedia returns a fresh signed URL for media/file particles. func (h *Handler) DownloadParticleMedia(w http.ResponseWriter, r *http.Request) { _, ok := middleware.EmailFromContext(r.Context()) if !ok { @@ -759,7 +742,7 @@ func (h *Handler) DownloadParticleMedia(w http.ResponseWriter, r *http.Request) // Depot Handlers // ============================================================================ -// PrepareUpload prepares an upload and returns a signed URL for direct upload to GCS +// PrepareUpload returns a signed URL for direct upload to GCS. func (h *Handler) PrepareUpload(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -817,7 +800,6 @@ func (h *Handler) PrepareUpload(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// ConfirmUpload confirms that an upload has been completed func (h *Handler) ConfirmUpload(w http.ResponseWriter, r *http.Request) { _, ok := middleware.EmailFromContext(r.Context()) if !ok { @@ -885,7 +867,7 @@ type InviteWaitlistEntrantRequest struct { // Waitlist Handlers // ============================================================================ -// AddToWaitlist adds an email to the waitlist (public, no auth) +// AddToWaitlist is public — no auth required. func (h *Handler) AddToWaitlist(w http.ResponseWriter, r *http.Request) { var req AddToWaitlistRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -912,7 +894,7 @@ func (h *Handler) AddToWaitlist(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusCreated) } -// GetWaitlist returns all waitlist entries (admin-only) +// GetWaitlist is admin-only. func (h *Handler) GetWaitlist(w http.ResponseWriter, r *http.Request) { if !middleware.IsAdminFromContext(r.Context()) { http.Error(w, "forbidden", http.StatusForbidden) @@ -944,7 +926,7 @@ func (h *Handler) GetWaitlist(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } -// GetWaitlistEntry returns a single waitlist entry by email (admin-only) +// GetWaitlistEntry is admin-only. func (h *Handler) GetWaitlistEntry(w http.ResponseWriter, r *http.Request) { if !middleware.IsAdminFromContext(r.Context()) { http.Error(w, "forbidden", http.StatusForbidden) @@ -972,7 +954,7 @@ func (h *Handler) GetWaitlistEntry(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(waitlistEntryToDTO(entry)) } -// InviteWaitlistEntrant marks a waitlist entry as invited (admin-only) +// InviteWaitlistEntrant is admin-only. func (h *Handler) InviteWaitlistEntrant(w http.ResponseWriter, r *http.Request) { if !middleware.IsAdminFromContext(r.Context()) { http.Error(w, "forbidden", http.StatusForbidden) @@ -1083,7 +1065,7 @@ func (h *Handler) GetLivekitToken(w http.ResponseWriter, r *http.Request) { return } - // Compose room name encoding both network and stream IDs for webhook resolution + // Encode both IDs in the room name so the webhook handler can resolve them. roomName := req.NetworkId + "/" + req.StreamId token, err := h.livekitClient.GetJoinToken(roomName, humanId, humanEmail) @@ -1097,9 +1079,8 @@ func (h *Handler) GetLivekitToken(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(GetLivekitTokenResponse{Token: token, ServerUrl: h.livekitClient.ServerUrl()}) } -// HandleLivekitWebhook processes LiveKit webhook events for huddle presence. -// It verifies the webhook signature (not user auth), then updates the stream -// particle's huddle_active_participants field in Firestore. +// HandleLivekitWebhook verifies the webhook signature (not user auth) and +// reconciles huddle_active_participants on the stream particle in Firestore. func (h *Handler) HandleLivekitWebhook(w http.ResponseWriter, r *http.Request) { event, err := webhook.ReceiveWebhookEvent(r, h.livekitClient.KeyProvider()) if err != nil { @@ -1113,13 +1094,12 @@ func (h *Handler) HandleLivekitWebhook(w http.ResponseWriter, r *http.Request) { switch eventType { case "participant_joined", "participant_left", "room_finished": - // Handle these events + // fall through default: w.WriteHeader(http.StatusOK) return } - // Parse room name to extract networkId and streamId roomName := event.GetRoom().GetName() parts := strings.SplitN(roomName, "/", 2) if len(parts) != 2 { @@ -1135,14 +1115,13 @@ func (h *Handler) HandleLivekitWebhook(w http.ResponseWriter, r *http.Request) { var participantIds []string if eventType == "room_finished" { - // Room is done — clear the participants participantIds = []string{} } else { - // Use ListParticipants for authoritative state (avoids drift from missed webhooks) + // Authoritative list avoids drift from missed/out-of-order webhooks. participants, err := h.livekitClient.ListParticipants(ctx, roomName) if err != nil { slog.Error("failed to list participants", "error", err, "room", roomName) - // Return 200 so LiveKit doesn't retry + // 200 to suppress LiveKit retries. w.WriteHeader(http.StatusOK) return } diff --git a/go/internal/handler/push_token.go b/go/internal/handler/push_token.go index 0c07b1e..edff810 100644 --- a/go/internal/handler/push_token.go +++ b/go/internal/handler/push_token.go @@ -20,9 +20,8 @@ type UnregisterPushTokenRequest struct { Token string `json:"token"` } -// RegisterPushToken upserts an Expo push token against the authenticated human. -// Re-binding a token to a new human (e.g., after a device-level account switch) -// happens transparently via ON CONFLICT. +// RegisterPushToken upserts an Expo token; ON CONFLICT transparently re-binds +// a token to a new human after a device-level account switch. func (h *Handler) RegisterPushToken(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { @@ -54,8 +53,7 @@ func (h *Handler) RegisterPushToken(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } -// UnregisterPushToken removes a push token belonging to the authenticated human. -// Returns 204 even if the token wasn't found — idempotent from the client's POV. +// UnregisterPushToken returns 204 whether or not the token existed (idempotent). func (h *Handler) UnregisterPushToken(w http.ResponseWriter, r *http.Request) { humanId, ok := middleware.HumanIdFromContext(r.Context()) if !ok { diff --git a/go/internal/human/pushnotify/expo.go b/go/internal/human/pushnotify/expo.go index 546f7b6..62dc0dd 100644 --- a/go/internal/human/pushnotify/expo.go +++ b/go/internal/human/pushnotify/expo.go @@ -20,7 +20,7 @@ const ( ExpoErrorDeviceNotRegistered = "DeviceNotRegistered" ) -// Message is one push to one device. Sound defaults to "default" when empty. +// Sound defaults to "default" when empty (set in Send). type Message struct { To string `json:"to"` Title string `json:"title,omitempty"` @@ -29,9 +29,8 @@ type Message struct { Sound string `json:"sound,omitempty"` } -// Ticket is the synchronous response Expo returns per message. Status is -// either "ok" or "error". On error, Details["error"] carries the code (e.g. -// "DeviceNotRegistered", "MessageTooBig", "InvalidCredentials"). +// Status is "ok" or "error". On error, Details["error"] carries the code +// (e.g. "DeviceNotRegistered", "MessageTooBig", "InvalidCredentials"). type Ticket struct { Status string `json:"status"` ID string `json:"id,omitempty"` @@ -39,8 +38,8 @@ type Ticket struct { Details map[string]any `json:"details,omitempty"` } -// ExpoClient is a minimal HTTP client for the Expo Push API. It does NOT poll -// receipts and does NOT retry. +// ExpoClient does NOT poll receipts and does NOT retry — fire-and-forget, +// with DeviceNotRegistered handled out-of-band by the notifier. type ExpoClient struct { http *http.Client accessToken string @@ -54,14 +53,13 @@ func NewExpoClient(accessToken string) *ExpoClient { } type expoSendResponse struct { - Data []Ticket `json:"data"` - Errors []map[string]any `json:"errors,omitempty"` + Data []Ticket `json:"data"` + Errors []map[string]any `json:"errors,omitempty"` } -// Send delivers messages in batches of up to expoMaxBatchSize. Returned tickets -// preserve the input order across batches: tickets[i] corresponds to msgs[i]. -// A request-level error (network, 5xx, malformed body) aborts the remaining -// batches and is returned to the caller along with whatever tickets succeeded. +// Send batches msgs (cap expoMaxBatchSize) and preserves input order: +// tickets[i] corresponds to msgs[i]. A request-level failure aborts the +// remaining batches; tickets already collected are returned with the error. func (c *ExpoClient) Send(ctx context.Context, msgs []Message) ([]Ticket, error) { if len(msgs) == 0 { return nil, nil diff --git a/go/internal/human/pushnotify/notifier.go b/go/internal/human/pushnotify/notifier.go index 913ff60..a06c59a 100644 --- a/go/internal/human/pushnotify/notifier.go +++ b/go/internal/human/pushnotify/notifier.go @@ -11,14 +11,12 @@ import ( "google.golang.org/grpc" ) -// PusherClient is the subset of the pusher gRPC service the notifier needs. -// Defined here (rather than depending on the generated client interface) so -// tests can supply a fake without standing up a gRPC server. +// PusherClient is a narrow subset of the pusher gRPC service so tests can +// supply a fake without standing up a real server. type PusherClient interface { IsOnline(ctx context.Context, in *pbpusher.IsOnlineRequest, opts ...grpc.CallOption) (*pbpusher.IsOnlineResponse, error) } -// NotifyInput is everything the notifier needs to know about a single newly-created particle. type NotifyInput struct { NetworkID string SenderHumanID string @@ -39,12 +37,11 @@ type NotifyInput struct { Body string } -// Notifier orchestrates the per-particle fanout: -// 1. Resolve recipients (stream visibility ∩ network members, minus sender). -// 2. Filter out anyone with an active WebSocket connection. -// 3. Look up each remaining human's push tokens. -// 4. POST a single batched request to Expo. -// 5. Delete any token Expo reports as DeviceNotRegistered. +// Notifier fans out one particle to Expo: +// 1. Resolve recipients (visibility ∩ network members, minus sender). +// 2. Drop anyone currently connected via WebSocket. +// 3. Send a batched Expo request for the remainder's tokens. +// 4. Prune tokens Expo reports as DeviceNotRegistered. type Notifier struct { networkR network.Reader tokens Service @@ -127,9 +124,8 @@ func (n *Notifier) queryOnline(ctx context.Context, humanIDs []string) (map[stri return resp.Online, nil } -// cleanupDeadTokens DELETEs any token Expo reports as DeviceNotRegistered. -// This is the one feedback signal we honor — other ticket errors (e.g. -// MessageTooBig, RateLimit) are logged but never retried. +// DeviceNotRegistered is the one feedback signal we honor; other ticket +// errors (MessageTooBig, RateLimit, …) are logged and dropped. func (n *Notifier) cleanupDeadTokens(ctx context.Context, msgs []Message, tickets []Ticket) { for i, t := range tickets { if i >= len(msgs) { @@ -160,12 +156,12 @@ func buildMessages(tokens []*PushToken, in NotifyInput) []Message { } data := map[string]any{ - "kind": "particle_created", - "network_id": in.NetworkID, - "stream_id": in.StreamID, - "particle_id": in.ParticleID, - "sender_human_id": in.SenderHumanID, - "particle_kind": in.ParticleKind, + "kind": "particle_created", + "network_id": in.NetworkID, + "stream_id": in.StreamID, + "particle_id": in.ParticleID, + "sender_human_id": in.SenderHumanID, + "particle_kind": in.ParticleKind, } msgs := make([]Message, 0, len(tokens)) diff --git a/go/internal/human/pushnotify/service.go b/go/internal/human/pushnotify/service.go index 3f7f2e0..d049fdb 100644 --- a/go/internal/human/pushnotify/service.go +++ b/go/internal/human/pushnotify/service.go @@ -6,22 +6,18 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -// Service is the full surface for per-device push token storage. HTTP handlers -// use Register/Unregister; the worker's notifier uses ListForHumans and -// DeleteByToken. Both consumers share the same underlying repository. +// Service stores per-device Expo push tokens and exposes the operations +// needed by both the HTTP handlers and the worker-side notifier. type Service interface { - // Register upserts a token for the given human. Returns ErrInvalidToken / - // ErrInvalidPlatform on bad input. + // Register returns ErrInvalidToken / ErrInvalidPlatform on bad input. Register(ctx context.Context, humanID string, in RegisterInput) error - // Unregister removes a token, scoped to the calling human so a user can't - // delete another user's token. Returns ErrNotFound if the token doesn't - // belong to humanID (or doesn't exist). + // Unregister is scoped to humanID so a user can't delete another user's + // token. Returns ErrNotFound if the token isn't owned by humanID. Unregister(ctx context.Context, humanID, token string) error - // ListForHumans returns every push token belonging to any of the given - // human IDs. Returns an empty slice when nothing matches. + // ListForHumans returns an empty slice when nothing matches. ListForHumans(ctx context.Context, humanIDs []string) ([]*PushToken, error) - // DeleteByToken removes a token regardless of owning human. Used by the - // notifier to clean up after Expo returns DeviceNotRegistered. + // DeleteByToken removes a token regardless of owner — used to prune after + // Expo reports DeviceNotRegistered. DeleteByToken(ctx context.Context, token string) error } diff --git a/go/internal/human/pushnotify/token.go b/go/internal/human/pushnotify/token.go index 8a2d126..0bdc0f3 100644 --- a/go/internal/human/pushnotify/token.go +++ b/go/internal/human/pushnotify/token.go @@ -21,12 +21,12 @@ func (p Platform) Valid() bool { } type PushToken struct { - Token string - HumanID string - Platform Platform - AppVersion string - CreatedAt time.Time - LastSeenAt time.Time + Token string + HumanID string + Platform Platform + AppVersion string + CreatedAt time.Time + LastSeenAt time.Time } var ( diff --git a/go/internal/human/service.go b/go/internal/human/service.go index 211805c..2180443 100644 --- a/go/internal/human/service.go +++ b/go/internal/human/service.go @@ -13,15 +13,12 @@ var ErrNotFound = errors.New("human not found") type Service interface { GetOrCreateByEmail(ctx context.Context, email string) (*Human, error) - // GetByEmail returns ErrNotFound if no human found + // GetByEmail returns ErrNotFound if no human found. GetByEmail(ctx context.Context, email string) (*Human, error) - // GetByID returns ErrNotFound if no human found + // GetByID returns ErrNotFound if no human found. GetByID(ctx context.Context, id string) (*Human, error) - // ListAll returns all humans ListAll(ctx context.Context) ([]*Human, error) - // UpdateEmailNotificationsEnabled toggles email notification preference UpdateEmailNotificationsEnabled(ctx context.Context, id string, enabled bool) error - // UpdateLastEmailNotificationSentAt records when the last notification email was sent UpdateLastEmailNotificationSentAt(ctx context.Context, id string, t time.Time) error } diff --git a/go/internal/livekit/client.go b/go/internal/livekit/client.go index 8cce5eb..1d278b9 100644 --- a/go/internal/livekit/client.go +++ b/go/internal/livekit/client.go @@ -11,13 +11,11 @@ import ( ) type Client interface { - // GetJoinToken generates a JWT for a participant to join a room. - // Name will show up in the participant data. + // GetJoinToken mints a participant JWT; name surfaces as the display name. GetJoinToken(roomId string, humanId string, name string) (string, error) ServerUrl() string - // ListParticipants returns the current participants in a room. ListParticipants(ctx context.Context, roomName string) ([]*livekit.ParticipantInfo, error) - // KeyProvider returns the key provider for verifying webhook signatures. + // KeyProvider is used by handlers to verify webhook signatures. KeyProvider() auth.KeyProvider } diff --git a/go/internal/livestore/membershippublisher.go b/go/internal/livestore/membershippublisher.go index a9d47f3..e4c2f72 100644 --- a/go/internal/livestore/membershippublisher.go +++ b/go/internal/livestore/membershippublisher.go @@ -8,10 +8,9 @@ import ( //go:generate go tool mockgen -source ./membershippublisher.go -destination ./mocks/membershippublisher.go -// MembershipPublisher publishes network membership changes to the live store -// (Firestore) that clients subscribe to. Postgres remains the source of truth; -// the membership reconciler heals any drift, so callers may log and ignore -// publish failures. +// MembershipPublisher fans network membership changes out to Firestore. +// Postgres is the source of truth; the reconciler heals drift, so publish +// failures are safe to log and ignore. type MembershipPublisher interface { Add(ctx context.Context, humanId, networkID string) error Remove(ctx context.Context, humanId, networkID string) error diff --git a/go/internal/media/helpers.go b/go/internal/media/helpers.go index 08eb06b..d75c33c 100644 --- a/go/internal/media/helpers.go +++ b/go/internal/media/helpers.go @@ -30,17 +30,15 @@ type TranscodeInput struct { type TranscodeOutput struct { TempLocalFilePath string OutputMimeType string - // Extension such as ".m4a" or ".mp4" - OutputExt string + OutputExt string // e.g. ".m4a" or ".mp4" } var ( ErrInvalidInput error = errors.New("invalid input") ) -// TranscodeToMp4 takes in any audio or video source URL and -// returns the filepath of the transcoded media -// WARNING: caller responsible for deleting TempLocalFilePath +// TranscodeToMp4 writes the result to a temp file; caller is responsible +// for deleting TempLocalFilePath. func TranscodeToMp4(ctx context.Context, input TranscodeInput) (*TranscodeOutput, error) { if input.SourceURL == "" || input.MimeType == "" { return nil, ErrInvalidInput @@ -74,11 +72,9 @@ func TranscodeToMp4(ctx context.Context, input TranscodeInput) (*TranscodeOutput tmpPath, } } else { - // Cap encoder parallelism and lookahead to keep memory bounded — screen - // recordings come in at native display resolution (often 1440p–4K) and - // libx264's per-thread lookahead/reference buffers blow past the worker's - // memory limit otherwise. Output is also downscaled to 1080p max, which - // mobile playback won't notice; the original WebM stays in GCS untouched. + // 1440p–4K screen recordings + libx264's lookahead buffers can OOM the + // worker. Bound parallelism/lookahead and downscale to 1080p; the + // original WebM stays in GCS untouched. args = []string{ "-y", "-i", input.SourceURL, "-vf", "scale='min(1920,iw)':-2:flags=lanczos", diff --git a/go/internal/middleware/auth.go b/go/internal/middleware/auth.go index f061586..7dedfec 100644 --- a/go/internal/middleware/auth.go +++ b/go/internal/middleware/auth.go @@ -17,40 +17,35 @@ const ( isAdminContextKey contextKey = "isAdmin" ) -// WithEmail adds the email to the context func WithEmail(ctx context.Context, email string) context.Context { return context.WithValue(ctx, emailContextKey, email) } -// EmailFromContext extracts the email from the context func EmailFromContext(ctx context.Context) (string, bool) { email, ok := ctx.Value(emailContextKey).(string) return email, ok } -// WithHumanId adds the humanId to the context func WithHumanId(ctx context.Context, humanId string) context.Context { return context.WithValue(ctx, humanIdContextKey, humanId) } -// HumanIdFromContext extracts the id from the context func HumanIdFromContext(ctx context.Context) (string, bool) { humanId, ok := ctx.Value(humanIdContextKey).(string) return humanId, ok } -// WithIsAdmin adds the admin flag to the context func WithIsAdmin(ctx context.Context, isAdmin bool) context.Context { return context.WithValue(ctx, isAdminContextKey, isAdmin) } -// IsAdminFromContext extracts the admin flag from the context func IsAdminFromContext(ctx context.Context) bool { isAdmin, ok := ctx.Value(isAdminContextKey).(bool) return ok && isAdmin } -// Auth returns a middleware that validates the session token and adds the email to the context +// Auth validates the bearer session token and populates email/humanId/isAdmin +// into the request context for downstream handlers. func Auth(authSvc auth.AuthService) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -66,7 +61,6 @@ func Auth(authSvc auth.AuthService) func(http.Handler) http.Handler { return } - // Auto-extend session if err := authSvc.ExtendSession(r.Context(), token); err != nil { slog.Warn("failed to extend session", "error", err) } @@ -79,7 +73,6 @@ func Auth(authSvc auth.AuthService) func(http.Handler) http.Handler { } } -// extractBearerToken extracts the token from the Authorization header func extractBearerToken(r *http.Request) string { authHeader := r.Header.Get("Authorization") if authHeader == "" { diff --git a/go/internal/middleware/cors.go b/go/internal/middleware/cors.go index f790338..f435de5 100644 --- a/go/internal/middleware/cors.go +++ b/go/internal/middleware/cors.go @@ -2,7 +2,7 @@ package middleware import "net/http" -// CORS wraps a handler to add CORS headers and handle preflight requests. +// CORS adds CORS headers and short-circuits preflight requests. func CORS(allowedOrigins []string) func(http.Handler) http.Handler { originSet := make(map[string]struct{}, len(allowedOrigins)) for _, o := range allowedOrigins { @@ -13,7 +13,7 @@ func CORS(allowedOrigins []string) func(http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { origin := r.Header.Get("Origin") - // Check if the origin is allowed (empty allowedOrigins means allow all) + // Empty allowedOrigins means allow all. allowed := len(originSet) == 0 if !allowed { _, allowed = originSet[origin] @@ -27,7 +27,6 @@ func CORS(allowedOrigins []string) func(http.Handler) http.Handler { w.Header().Set("Access-Control-Max-Age", "86400") } - // Handle preflight if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) return diff --git a/go/internal/network/reader.go b/go/internal/network/reader.go index 9cfceec..8068c00 100644 --- a/go/internal/network/reader.go +++ b/go/internal/network/reader.go @@ -16,14 +16,11 @@ type Reader interface { ListForHuman(ctx context.Context, humanId string) ([]*Network, error) // IsMember returns ErrInvalidHumanId if humanId is empty. IsMember(ctx context.Context, networkID, humanId string) (bool, error) - // ListMembers returns all human IDs belonging to the network. - // Returns an empty slice if the network doesn't exist. + // ListMembers returns an empty slice if the network doesn't exist. ListMembers(ctx context.Context, networkID string) ([]string, error) - // ListAll returns all networks with their members ListAll(ctx context.Context) ([]*Network, error) // ListAllMemberships returns humanId -> networkIds for every human with at // least one membership. Humans with zero memberships are absent from the map. - // Used by the membership reconciler to diff the Firestore mirror. ListAllMemberships(ctx context.Context) (map[string][]string, error) CountSeats(ctx context.Context, networkID string) (int, error) @@ -37,8 +34,8 @@ type readerImpl struct { repo repository } -// newReader returns the concrete reader. Used by NewService to embed without -// going through the Reader interface (which would hide pool/repo). +// newReader exposes the concrete type so the service can embed it without +// hiding pool/repo behind the Reader interface. func newReader(pool *pgxpool.Pool) *readerImpl { return &readerImpl{ pool: pool, diff --git a/go/internal/network/repository.go b/go/internal/network/repository.go index f7a5544..78359a9 100644 --- a/go/internal/network/repository.go +++ b/go/internal/network/repository.go @@ -10,9 +10,8 @@ import ( "go.jetify.com/typeid" ) -// dbtx is the subset of pgx's query API shared by *pgxpool.Pool and pgx.Tx. -// Used by repository helpers that the service layer may run either standalone -// (against the pool) or inside a transaction. +// dbtx is the subset of pgx's query API shared by *pgxpool.Pool and pgx.Tx, +// so repository helpers can run standalone or inside a transaction. type dbtx interface { Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) @@ -57,8 +56,7 @@ type repository interface { deleteInvitation(ctx context.Context, db dbtx, networkID, email string) error } -// networkColumns lists every column selected when hydrating a Network. -// Centralized to keep SELECTs and Scan() calls in sync. +// Centralized so SELECTs and scanNetwork stay in sync. const networkColumns = `id, name, admin_human_id, created_at` func scanNetwork(row pgx.Row, n *Network) error { @@ -283,8 +281,6 @@ func (r *repositoryImpl) listAll(ctx context.Context) ([]*Network, error) { return networks, nil } -// Invitation methods - func (r *repositoryImpl) createInvitation(ctx context.Context, networkID, email string) error { _, err := r.pool.Exec(ctx, `INSERT INTO network_invitations (network_id, email) VALUES ($1, $2) diff --git a/go/internal/network/service.go b/go/internal/network/service.go index 8cacac9..734a7bd 100644 --- a/go/internal/network/service.go +++ b/go/internal/network/service.go @@ -28,7 +28,7 @@ var ErrInvalidRetentionHours = errors.New("message retention hours must be betwe type Service interface { Reader - // Create creates a network and adds adminHumanId as the first member. Returns ErrInvalidName if name is empty. + // Create adds adminHumanId as the first member. Returns ErrInvalidName if name is empty. Create(ctx context.Context, name, adminHumanId string) (*Network, error) // SetName returns ErrNotFound or ErrInvalidName. SetName(ctx context.Context, id, name string) error @@ -145,10 +145,8 @@ func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId strin return nil } -// mirrorAddMembership / mirrorRemoveMembership keep the live store membership -// projection (humans/{humanId}.networks) in sync with Postgres. Called after -// the Postgres transaction commits. Failures are logged but not returned: -// Postgres is the source of truth and the reconciler will heal drift. +// Mirror the live store membership projection (humans/{humanId}.networks). +// Postgres is the source of truth: failures are logged and the reconciler heals drift. func (s *serviceImpl) mirrorAddMembership(ctx context.Context, humanId, networkID string) { if err := s.pub.Add(ctx, humanId, networkID); err != nil { slog.Error("membership publish add failed", "error", err, "humanId", humanId, "networkID", networkID) @@ -161,8 +159,7 @@ func (s *serviceImpl) mirrorRemoveMembership(ctx context.Context, humanId, netwo } } -// mutateMembers runs fn in a tx, recounts seats, calls billing.SyncSeats, -// and commits. Any error rolls the membership change back. +// Runs fn in a tx and syncs seats to billing atomically. Any error rolls back. func (s *serviceImpl) mutateMembers(ctx context.Context, networkID string, fn func(pgx.Tx) error) error { tx, err := s.pool.Begin(ctx) if err != nil { diff --git a/go/internal/particle/models.go b/go/internal/particle/models.go index 26e6a1b..0263e4b 100644 --- a/go/internal/particle/models.go +++ b/go/internal/particle/models.go @@ -6,7 +6,6 @@ import ( "time" ) -// ParticleType represents the type of particle type ParticleType string const ( @@ -20,7 +19,6 @@ const ( // TypeThink ParticleType = "think" ) -// VisibilityMode represents how access to a particle is determined type VisibilityMode string const ( @@ -32,7 +30,6 @@ const ( var ErrInvalidParticleType = errors.New("invalid particle type") var ErrInvalidVisibilityMode = errors.New("invalid visibility mode") -// ParseParticleType parses a string into a ParticleType func ParseParticleType(s string) (ParticleType, error) { switch s { case string(TypeStream): @@ -54,7 +51,6 @@ func ParseParticleType(s string) (ParticleType, error) { } } -// ParseVisibilityMode parses a string into a VisibilityMode func ParseVisibilityMode(s string) (VisibilityMode, error) { switch s { case "", string(VisibilityNetworkAll): @@ -68,7 +64,6 @@ func ParseVisibilityMode(s string) (VisibilityMode, error) { } } -// Stream status values type StreamStatus string const ( @@ -76,7 +71,6 @@ const ( StreamStatusClosed StreamStatus = "closed" ) -// Particle represents a content particle in the system type Particle struct { ID string Type ParticleType @@ -89,7 +83,6 @@ type Particle struct { CreatedAt time.Time } -// CreateInput represents the input for creating a new particle type CreateInput struct { Type ParticleType NetworkID string @@ -99,18 +92,15 @@ type CreateInput struct { Visibility VisibilityMode } -// ListFilter represents filtering options for listing particles type ListFilter struct { Types []ParticleType } -// Cursor represents a pagination cursor for bidirectional pagination type Cursor struct { Position string // particle ID or timestamp Direction string // "before" or "after" } -// ParticleList represents a paginated list of particles type ParticleList struct { Particles []*Particle HasMore bool @@ -118,57 +108,48 @@ type ParticleList struct { PrevCursor *Cursor } -// StreamData represents the data stored for stream particles type StreamData struct { Name string `json:"name"` Status string `json:"status"` // "open" or "closed" Description *string `json:"description"` } -// FolderData represents the data stored for folder particles type FolderData struct { Name string `json:"name"` Color *string `json:"color"` } -// MediaData represents the data stored for media particles type MediaData struct { - ObjectID string `json:"object_id"` // reference to storage object + ObjectID string `json:"object_id"` MimeType string `json:"mime_type"` DurationMs int `json:"duration_ms"` - // Caption *string `json:"caption"` } -// FileData represents the data stored for file particles type FileData struct { - ObjectID string `json:"object_id"` // reference to storage object + ObjectID string `json:"object_id"` Filename string `json:"filename"` MimeType string `json:"mime_type"` - Size int64 `json:"size"` // in bytes + Size int64 `json:"size"` // bytes } -// TextData represents the data stored for text particles type TextData struct { Content string `json:"content"` } -// QuestData represents the data stored for quest particles type QuestData struct { Title string `json:"title"` Description string `json:"description"` Done bool `json:"done"` Status *string `json:"status"` AssignedTo *string `json:"assigned_to,omitempty"` // email - DueDate *string `json:"due_date,omitempty"` // ISO date string + DueDate *string `json:"due_date,omitempty"` // ISO date } -// PaperData represents the data stored for paper particles type PaperData struct { Title string `json:"title"` Content string `json:"content"` // markdown } -// AckInfo represents an acknowledgment record type AckInfo struct { Email string AckedAt time.Time diff --git a/go/internal/particle/networkmembershipchecker.go b/go/internal/particle/networkmembershipchecker.go index 7886d1e..bb317b5 100644 --- a/go/internal/particle/networkmembershipchecker.go +++ b/go/internal/particle/networkmembershipchecker.go @@ -5,6 +5,5 @@ import "context" //go:generate go tool mockgen -source ./networkmembershipchecker.go -destination ./mocks/networkmembershipchecker.go type NetworkMembershipChecker interface { - // IsMember returns true if the humanId is a member of the network. IsMember(ctx context.Context, networkID, humanId string) (bool, error) } diff --git a/go/internal/particle/repository.go b/go/internal/particle/repository.go index 7b8089f..8eee46d 100644 --- a/go/internal/particle/repository.go +++ b/go/internal/particle/repository.go @@ -40,7 +40,7 @@ type repository interface { getMembers(ctx context.Context, particleID string) ([]string, error) getMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error) - // getAncestorChain returns the particle and all its ancestors (for access checks) + // getAncestorChain returns the particle followed by its ancestors, in order. getAncestorChain(ctx context.Context, particleID string) ([]*Particle, error) isMemberOf(ctx context.Context, particleID, email string) (bool, error) diff --git a/go/internal/particle/service.go b/go/internal/particle/service.go index 80b2d92..0d1c882 100644 --- a/go/internal/particle/service.go +++ b/go/internal/particle/service.go @@ -13,45 +13,46 @@ import ( const defaultPageSize = 50 -// NOTE: this service is deprecated as we use firestore for particle data +// Deprecated: particle data now lives in Firestore. The Postgres-backed +// service is retained only for legacy paths. type Service interface { - // Create creates a new particle. Caller must be a network member (verified by handler). - // Returns ErrInvalidType, ErrInvalidData, ErrMembersRequired, ErrInvalidParent, ErrAccessDenied, or ErrCapacityExceeded. + // Create returns ErrInvalidType, ErrInvalidData, ErrMembersRequired, + // ErrInvalidParent, ErrAccessDenied, or ErrCapacityExceeded. Network + // membership is verified by the handler. Create(ctx context.Context, input CreateInput, requesterEmail string) (*Particle, error) // GetByID returns ErrNotFound or ErrAccessDenied. GetByID(ctx context.Context, id, requesterEmail string) (*Particle, error) - // Update updates the particle's data. Returns ErrNotFound, ErrAccessDenied, or ErrInvalidData. + // Update returns ErrNotFound, ErrAccessDenied, or ErrInvalidData. Update(ctx context.Context, id string, data json.RawMessage, requesterEmail string) (*Particle, error) // Delete returns ErrNotFound or ErrAccessDenied. Delete(ctx context.Context, id, requesterEmail string) error - // List returns particles in a network. Use parentID=nil for root particles. - // Returns ErrNotFound or ErrAccessDenied if parentID is specified and inaccessible. + // List uses parentID=nil for root particles. Returns ErrNotFound or + // ErrAccessDenied when parentID is given but inaccessible. List(ctx context.Context, networkID string, parentID *string, requesterEmail string, filter ListFilter, cursor *Cursor, limit int) (*ParticleList, error) - // OpenStream opens a closed stream. Returns ErrNotFound, ErrAccessDenied, ErrNotAStream, ErrStreamAlreadyOpen, or ErrCapacityExceeded. + // OpenStream returns ErrNotFound, ErrAccessDenied, ErrNotAStream, + // ErrStreamAlreadyOpen, or ErrCapacityExceeded. OpenStream(ctx context.Context, id, requesterEmail string) error - // CloseStream closes an open stream. Returns ErrNotFound, ErrAccessDenied, ErrNotAStream, or ErrStreamAlreadyClosed. + // CloseStream returns ErrNotFound, ErrAccessDenied, ErrNotAStream, or + // ErrStreamAlreadyClosed. CloseStream(ctx context.Context, id, requesterEmail string) error - // SetVisibility changes the particle's visibility mode. Returns ErrNotFound, ErrAccessDenied, or ErrAccessExpansion. + // SetVisibility returns ErrNotFound, ErrAccessDenied, or ErrAccessExpansion. SetVisibility(ctx context.Context, id string, mode VisibilityMode, requesterEmail string) error - // AddMembers adds members to a custom visibility particle. Returns ErrNotFound or ErrAccessDenied. + // AddMembers / RemoveMembers operate on custom-visibility streams only. + // Both return ErrNotFound or ErrAccessDenied. AddMembers(ctx context.Context, id string, emails []string, requesterEmail string) error - // RemoveMembers removes members from a custom visibility particle. Returns ErrNotFound or ErrAccessDenied. RemoveMembers(ctx context.Context, id string, emails []string, requesterEmail string) error - // Seen tracking (private) + // Seen tracking is private per human; Ack is public and permanent. MarkSeen(ctx context.Context, id, requesterEmail string) error MarkSeenBatch(ctx context.Context, ids []string, requesterEmail string) error - - // Ack tracking (public, permanent) Ack(ctx context.Context, id, requesterEmail string) error - // Unseen counts for stream list view GetUnseenCounts(ctx context.Context, networkID string, streamIDs []string, requesterEmail string) (map[string]int, error) - // Bulk lookups for handler enrichment + // Bulk lookups for batch hydration. GetSeenMap(ctx context.Context, particleIDs []string, requesterEmail string) (map[string]bool, error) GetAcksMap(ctx context.Context, particleIDs []string) (map[string][]AckInfo, error) GetMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error) @@ -69,10 +70,9 @@ func NewService(pool *pgxpool.Pool, networkReader NetworkMembershipChecker) Serv } } -// checkAccess verifies that the email has access to the particle based on visibility. -// Assumes the caller is already verified as a network member (handler responsibility). -// Walks up the ancestor chain only when visibility is inherited, stopping at the first -// network_all or custom node. +// Walks the ancestor chain when visibility is inherited, stopping at the +// first network_all or custom node. Assumes network membership is already +// verified by the handler. func (s *serviceImpl) checkAccess(ctx context.Context, particleID, email string) (bool, error) { ancestors, err := s.repo.getAncestorChain(ctx, particleID) if err != nil { @@ -83,13 +83,12 @@ func (s *serviceImpl) checkAccess(ctx context.Context, particleID, email string) return false, errNotFound } - // Build lookup map by ID byID := make(map[string]*Particle, len(ancestors)) for _, p := range ancestors { byID[p.ID] = p } - // Start from the target particle (first in chain) and walk up on inherited + // ancestors[0] is the target; walk up only on inherited. current := ancestors[0] for { switch current.Visibility { @@ -99,7 +98,7 @@ func (s *serviceImpl) checkAccess(ctx context.Context, particleID, email string) return s.repo.isMemberOf(ctx, current.ID, email) case VisibilityInherited: if current.ParentID == nil { - // inherited at root is invalid state, deny access + // inherited-at-root is invalid; deny. return false, nil } parent, ok := byID[*current.ParentID] @@ -119,30 +118,24 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm return nil, err } - // Validate particle type if !isValidParticleType(input.Type) { return nil, ErrInvalidType } - // Validate data matches type requirements if err := validateParticleData(input.Type, input.Data); err != nil { return nil, err } - // MVP visibility rules: - // - Child particles (have parent) → always inherited - // - Root particles (no parent) → cannot be inherited, default network_all + // MVP visibility: children always inherit; roots cannot inherit and + // default to network_all. Streams/folders are root-only. if input.ParentID != nil { - // Children always inherit from parent input.Visibility = VisibilityInherited - input.Members = nil // no members on inherited particles + input.Members = nil - // Reject streams and folders as children (MVP: streams are root-level only) if input.Type == TypeStream || input.Type == TypeFolder { return nil, ErrInvalidParent } } else { - // Root particles cannot be inherited if input.Visibility == VisibilityInherited { return nil, ErrInheritedAtRoot } @@ -151,7 +144,6 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm } } - // Custom visibility requires at least one member and must be a stream var customMembers []string if input.Visibility == VisibilityCustom { if input.Type != TypeStream { @@ -161,7 +153,7 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm return nil, ErrMembersRequired } - // Validate every supplied member against the network checker before touching the DB. + // Validate every member upfront so DB writes are all-or-nothing. customMembers = make([]string, 0, len(input.Members)+1) customMembers = append(customMembers, requesterEmail) seen := map[string]bool{requesterEmail: true} @@ -186,8 +178,7 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm } } - // Network membership is verified by handler - we only check particle visibility - // If parent specified, check parent access (visibility-based) + // Network membership is verified by the handler; only particle visibility is checked here. if input.ParentID != nil { hasAccess, err := s.checkAccess(ctx, *input.ParentID, requesterEmail) if err != nil { @@ -201,7 +192,6 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm } } - // Build the particle p := &Particle{ Type: input.Type, NetworkID: input.NetworkID, @@ -215,9 +205,8 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm p.Data = json.RawMessage("{}") } - // For streams, set initial status to open and check capacity + // New streams default to open. if input.Type == TypeStream { - // Set status to open in the data JSON data, err := setStreamStatus(p.Data, string(StreamStatusOpen)) if err != nil { return nil, err @@ -225,13 +214,11 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm p.Data = data } - // Create the particle created, err := s.repo.create(ctx, p) if err != nil { return nil, err } - // Add the pre-validated member list for custom visibility. if len(customMembers) > 0 { if err := s.repo.addMembers(ctx, created.ID, customMembers); err != nil { return nil, err @@ -247,7 +234,6 @@ func (s *serviceImpl) GetByID(ctx context.Context, id, requesterEmail string) (* return nil, err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -275,7 +261,6 @@ func (s *serviceImpl) Update(ctx context.Context, id string, data json.RawMessag return nil, err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -287,7 +272,6 @@ func (s *serviceImpl) Update(ctx context.Context, id string, data json.RawMessag return nil, ErrAccessDenied } - // Get the particle to validate data against its type p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -296,7 +280,6 @@ func (s *serviceImpl) Update(ctx context.Context, id string, data json.RawMessag return nil, err } - // Validate data matches type requirements if err := validateParticleData(p.Type, data); err != nil { return nil, err } @@ -318,7 +301,6 @@ func (s *serviceImpl) Delete(ctx context.Context, id, requesterEmail string) err return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -330,7 +312,6 @@ func (s *serviceImpl) Delete(ctx context.Context, id, requesterEmail string) err return ErrAccessDenied } - // Get the particle to check if it's an open stream _, err = s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -352,8 +333,7 @@ func (s *serviceImpl) List(ctx context.Context, networkID string, parentID *stri return nil, err } - // Network membership is verified by handler - we only check particle visibility - // If parentID specified, check access to parent (visibility-based) + // Network membership is verified by the handler; only particle visibility is checked here. if parentID != nil { hasAccess, err := s.checkAccess(ctx, *parentID, requesterEmail) if err != nil { @@ -367,8 +347,7 @@ func (s *serviceImpl) List(ctx context.Context, networkID string, parentID *stri } } - // Fetch one extra to determine if there are more - // Access filtering is done in the query itself (network_all OR user is member) + // Fetch limit+1 to detect a next page; visibility filtering lives in the query. if limit == 0 { limit = defaultPageSize } @@ -415,7 +394,6 @@ func (s *serviceImpl) OpenStream(ctx context.Context, id, requesterEmail string) return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -427,7 +405,6 @@ func (s *serviceImpl) OpenStream(ctx context.Context, id, requesterEmail string) return ErrAccessDenied } - // Get the particle p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -444,7 +421,6 @@ func (s *serviceImpl) OpenStream(ctx context.Context, id, requesterEmail string) return ErrStreamAlreadyOpen } - // Update stream status in data newData, err := setStreamStatus(p.Data, string(StreamStatusOpen)) if err != nil { return err @@ -467,7 +443,6 @@ func (s *serviceImpl) CloseStream(ctx context.Context, id, requesterEmail string return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -479,7 +454,6 @@ func (s *serviceImpl) CloseStream(ctx context.Context, id, requesterEmail string return ErrAccessDenied } - // Get the particle p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -496,7 +470,6 @@ func (s *serviceImpl) CloseStream(ctx context.Context, id, requesterEmail string return ErrStreamAlreadyClosed } - // Update stream status in data newData, err := setStreamStatus(p.Data, string(StreamStatusClosed)) if err != nil { return err @@ -519,7 +492,6 @@ func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode Visibil return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -531,7 +503,6 @@ func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode Visibil return ErrAccessDenied } - // Get the particle to check constraints p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -540,12 +511,11 @@ func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode Visibil return err } - // Root particles cannot be inherited if mode == VisibilityInherited && p.ParentID == nil { return ErrInheritedAtRoot } - // If expanding to network_all, check that parent's effective visibility allows it + // Expanding to network_all is rejected if any ancestor restricts to custom. if mode == VisibilityNetworkAll && p.ParentID != nil { parentVis, err := s.getEffectiveVisibility(ctx, *p.ParentID) if err != nil { @@ -563,7 +533,7 @@ func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode Visibil return err } -// getEffectiveVisibility walks up the inherited chain to find the concrete visibility mode. +// Walks up the inherited chain to the concrete visibility node. func (s *serviceImpl) getEffectiveVisibility(ctx context.Context, particleID string) (VisibilityMode, error) { ancestors, err := s.repo.getAncestorChain(ctx, particleID) if err != nil { @@ -600,7 +570,6 @@ func (s *serviceImpl) AddMembers(ctx context.Context, id string, emails []string return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -612,7 +581,6 @@ func (s *serviceImpl) AddMembers(ctx context.Context, id string, emails []string return ErrAccessDenied } - // Get the particle to check type and parent access p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -621,14 +589,11 @@ func (s *serviceImpl) AddMembers(ctx context.Context, id string, emails []string return err } - // Only streams can have members if p.Type != TypeStream { return ErrNotAContainer } - // Validate and normalize emails, checking network membership upfront. - // Strict: a normalize failure, checker error, or non-member aborts the - // whole operation before any rows are written. + // Validate every email upfront so any failure aborts before DB writes. normalizedEmails := make([]string, 0, len(emails)) seen := make(map[string]bool, len(emails)) for _, email := range emails { @@ -664,7 +629,6 @@ func (s *serviceImpl) RemoveMembers(ctx context.Context, id string, emails []str return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -676,7 +640,6 @@ func (s *serviceImpl) RemoveMembers(ctx context.Context, id string, emails []str return ErrAccessDenied } - // Get the particle to check type p, err := s.repo.getByID(ctx, id) if err != nil { if errors.Is(err, errNotFound) { @@ -685,7 +648,6 @@ func (s *serviceImpl) RemoveMembers(ctx context.Context, id string, emails []str return err } - // Only streams can have members if p.Type != TypeStream { return ErrNotAContainer } @@ -712,7 +674,6 @@ func (s *serviceImpl) MarkSeen(ctx context.Context, id, requesterEmail string) e return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -733,17 +694,17 @@ func (s *serviceImpl) MarkSeenBatch(ctx context.Context, ids []string, requester return err } - // Check access for each particle and mark seen + // Silently skip particles that are missing or inaccessible. for _, id := range ids { hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { - continue // Skip non-existent particles + continue } return err } if !hasAccess { - continue // Skip inaccessible particles + continue } if err := s.repo.markSeen(ctx, id, requesterEmail); err != nil { @@ -760,7 +721,6 @@ func (s *serviceImpl) Ack(ctx context.Context, id, requesterEmail string) error return err } - // Check access hasAccess, err := s.checkAccess(ctx, id, requesterEmail) if err != nil { if errors.Is(err, errNotFound) { @@ -772,7 +732,7 @@ func (s *serviceImpl) Ack(ctx context.Context, id, requesterEmail string) error return ErrAccessDenied } - // Ack also marks as seen + // Ack implies seen. if err := s.repo.markSeen(ctx, id, requesterEmail); err != nil { return err } @@ -815,7 +775,6 @@ func isValidParticleType(t ParticleType) bool { } } -// getStreamStatus extracts the status from a stream particle's data func getStreamStatus(data json.RawMessage) string { var d StreamData if err := json.Unmarshal(data, &d); err != nil { @@ -824,7 +783,6 @@ func getStreamStatus(data json.RawMessage) string { return d.Status } -// setStreamStatus updates the status in a stream particle's data func setStreamStatus(data json.RawMessage, status string) (json.RawMessage, error) { var d StreamData if err := json.Unmarshal(data, &d); err != nil { @@ -834,10 +792,9 @@ func setStreamStatus(data json.RawMessage, status string) (json.RawMessage, erro return json.Marshal(d) } -// validateParticleData validates that the data field contains valid JSON -// and has required fields for the given particle type. +// Returns ErrInvalidData if data is not valid JSON or is missing required +// fields for pType. Empty/null data is allowed and treated as {}. func validateParticleData(pType ParticleType, data json.RawMessage) error { - // Empty or null data is allowed - will default to {} if len(data) == 0 || string(data) == "null" || string(data) == "{}" { return nil } diff --git a/go/internal/pusher/authorizer.go b/go/internal/pusher/authorizer.go index a67f4c6..62e8040 100644 --- a/go/internal/pusher/authorizer.go +++ b/go/internal/pusher/authorizer.go @@ -10,20 +10,19 @@ import ( var ErrUnauthorized = errors.New("unauthorized") -// Authorizer validates whether a user can access a given channel. type Authorizer struct { networkReader network.Reader } -// NewAuthorizer creates a new channel authorizer. func NewAuthorizer(networkReader network.Reader) *Authorizer { return &Authorizer{networkReader: networkReader} } -// Authorize checks if the given humanID is allowed to subscribe to the channel. -// Channel formats: -// - network:{networkId} -// - stream:{networkId}:{streamId} +// Authorize accepts channel IDs of the form: +// +// network:{networkId} +// stream:{networkId}:{streamId} +// _presence:{humanId} func (a *Authorizer) Authorize(ctx context.Context, channelID, humanID string) error { parts := strings.SplitN(channelID, ":", 2) if len(parts) < 2 { @@ -39,8 +38,7 @@ func (a *Authorizer) Authorize(ctx context.Context, channelID, humanID string) e case "stream": return a.authorizeStream(ctx, rest, humanID) case "_presence": - // Always allowed — used for global online presence tracking. - // The channel ID is _presence:{humanId}, so verify the humanId matches. + // Only the owning human may subscribe to their presence channel. if rest != humanID { return ErrUnauthorized } @@ -61,8 +59,7 @@ func (a *Authorizer) authorizeNetwork(ctx context.Context, networkID, humanID st return nil } -// authorizeStream expects rest to be "{networkId}:{streamId}". -// We only check network membership — stream visibility is handled by network access. +// rest is "{networkId}:{streamId}"; stream-level visibility is enforced by network access. func (a *Authorizer) authorizeStream(ctx context.Context, rest, humanID string) error { parts := strings.SplitN(rest, ":", 2) if len(parts) < 2 { diff --git a/go/internal/pusher/channel.go b/go/internal/pusher/channel.go index e1025c1..054f3f6 100644 --- a/go/internal/pusher/channel.go +++ b/go/internal/pusher/channel.go @@ -1,7 +1,7 @@ package pusher -// Channel tracks the local connections subscribed to a channel on this pod. -// All methods are only called from the Hub goroutine — no locks needed. +// Channel tracks the local connections subscribed on this pod. +// State is only mutated by the Hub goroutine, so no locks are needed. type Channel struct { id string members map[*Conn]string // conn → humanID @@ -26,7 +26,7 @@ func (ch *Channel) isEmpty() bool { return len(ch.members) == 0 } -// localHumanIDs returns the deduplicated set of humanIDs connected on this pod. +// Deduplicated set; the same human may have multiple connections. func (ch *Channel) localHumanIDs() []string { seen := make(map[string]bool, len(ch.members)) ids := make([]string, 0, len(ch.members)) @@ -39,7 +39,6 @@ func (ch *Channel) localHumanIDs() []string { return ids } -// hasHumanID returns true if the given humanID has at least one local connection. func (ch *Channel) hasHumanID(humanID string) bool { for _, hid := range ch.members { if hid == humanID { @@ -49,7 +48,6 @@ func (ch *Channel) hasHumanID(humanID string) bool { return false } -// broadcast sends a message to all local connections except the excluded one. func (ch *Channel) broadcast(msg ServerMessage, exclude *Conn) { for conn := range ch.members { if conn != exclude { diff --git a/go/internal/pusher/conn.go b/go/internal/pusher/conn.go index 3cdf2c0..6a73912 100644 --- a/go/internal/pusher/conn.go +++ b/go/internal/pusher/conn.go @@ -11,13 +11,12 @@ import ( const sendBufferSize = 256 -// Conn wraps a WebSocket connection with identity and a send buffer. type Conn struct { id string humanID string ws *websocket.Conn send chan []byte - once sync.Once // ensures close logic runs once + once sync.Once // guards Close } func newConn(id, humanID string, ws *websocket.Conn) *Conn { @@ -29,8 +28,8 @@ func newConn(id, humanID string, ws *websocket.Conn) *Conn { } } -// ReadPump reads messages from the WebSocket and forwards them to the hub. -// It blocks until the connection is closed or the context is cancelled. +// ReadPump forwards inbound frames to the hub; blocks until the connection +// closes or ctx is cancelled. func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { defer hub.disconnect(c) @@ -45,7 +44,6 @@ func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { return } - // Respond to keep-alive pings if string(data) == "ping" { if err := c.ws.Write(ctx, websocket.MessageText, []byte("pong")); err != nil { slog.Warn("websocket pong write error", "connId", c.id, "error", err) @@ -84,7 +82,6 @@ func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { } } -// WritePump drains the send buffer and writes messages to the WebSocket. func (c *Conn) WritePump(ctx context.Context) { for { select { @@ -102,8 +99,7 @@ func (c *Conn) WritePump(ctx context.Context) { } } -// Send enqueues a ServerMessage to be written to the WebSocket. -// If the send buffer is full, the connection is closed (slow client). +// A full send buffer closes the connection (slow client policy). func (c *Conn) Send(msg ServerMessage) { data, err := json.Marshal(msg) if err != nil { @@ -119,7 +115,6 @@ func (c *Conn) Send(msg ServerMessage) { } } -// Close closes the WebSocket connection and the send channel. func (c *Conn) Close() { c.once.Do(func() { c.ws.Close(websocket.StatusNormalClosure, "closing") diff --git a/go/internal/pusher/hub.go b/go/internal/pusher/hub.go index 1de54e4..3ecc608 100644 --- a/go/internal/pusher/hub.go +++ b/go/internal/pusher/hub.go @@ -43,7 +43,6 @@ type Hub struct { remoteEventCh chan *remoteEvent } -// NewHub creates a new Hub. func NewHub(bridge *RedisBridge, authorizer *Authorizer) *Hub { return &Hub{ channels: make(map[string]*Channel), @@ -84,7 +83,6 @@ func (h *Hub) Run(ctx context.Context) { } func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { - // Authorize channel access if err := h.authorizer.Authorize(ctx, req.channelID, req.conn.humanID); err != nil { req.conn.Send(ServerMessage{ Type: TypeError, @@ -94,7 +92,6 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { return } - // Get or create local channel ch, ok := h.channels[req.channelID] if !ok { ch = newChannel(req.channelID) @@ -104,32 +101,27 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { // Capture before addMember so multi-tab joins don't emit a spurious join. wasPresentLocally := ch.hasHumanID(req.conn.humanID) - // Add to local channel ch.addMember(req.conn, req.conn.humanID) - // Track in reverse index if h.connChannels[req.conn] == nil { h.connChannels[req.conn] = make(map[string]bool) } h.connChannels[req.conn][req.channelID] = true - // Register in Redis and get global presence presence, err := h.bridge.Subscribe(ctx, req.channelID, req.conn.id, req.conn.humanID) if err != nil { slog.Error("redis subscribe failed", "channelId", req.channelID, "error", err) - // Still send local presence as fallback + // Fall back to local-only presence. presence = ch.localHumanIDs() } - // Send subscribed ack with presence snapshot req.conn.Send(ServerMessage{ Type: TypeSubscribed, Channel: req.channelID, Presence: presence, }) - // Notify other local members. The Redis self-filter drops our own echo, - // so same-pod peers would otherwise never hear about this join. + // Redis self-filter drops our own echo, so same-pod peers need a direct nudge. if !wasPresentLocally { ch.broadcast(ServerMessage{ Type: TypeJoin, @@ -147,18 +139,15 @@ func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { ch.removeMember(req.conn) - // Remove from reverse index if chans, ok := h.connChannels[req.conn]; ok { delete(chans, req.channelID) } - // Update Redis if err := h.bridge.Unsubscribe(ctx, req.channelID, req.conn.id, req.conn.humanID); err != nil { slog.Error("redis unsubscribe failed", "channelId", req.channelID, "error", err) } - // Notify other local members iff the humanID is fully gone from this pod - // (multi-tab: other conns keep them present, so no leave fires). + // Only emit leave once the humanID has no remaining tabs on this pod. if !ch.hasHumanID(req.conn.humanID) { ch.broadcast(ServerMessage{ Type: TypeLeave, @@ -167,7 +156,6 @@ func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { }, req.conn) } - // Clean up empty local channel if ch.isEmpty() { delete(h.channels, req.channelID) } @@ -179,13 +167,12 @@ func (h *Hub) handleBroadcast(ctx context.Context, req *broadcastRequest) { return } - // Check that the sender is actually in the channel if _, isMember := ch.members[req.conn]; !isMember { req.conn.sendError("not subscribed to channel: " + req.channelID) return } - // Deliver to local connections (except sender) + // Local fanout (excluding sender), then publish for other pods. ch.broadcast(ServerMessage{ Type: TypeMessage, Channel: req.channelID, @@ -193,7 +180,6 @@ func (h *Hub) handleBroadcast(ctx context.Context, req *broadcastRequest) { Payload: req.payload, }, req.conn) - // Publish to Redis for other pods h.bridge.Broadcast(ctx, req.channelID, req.conn.humanID, req.payload) } @@ -234,7 +220,8 @@ func (h *Hub) handleDisconnect(ctx context.Context, conn *Conn) { func (h *Hub) handleRemoteEvent(evt *remoteEvent) { ch, ok := h.channels[evt.channelID] if !ok { - return // no local connections care about this channel + // No local subscribers — drop the event. + return } switch evt.event.Type { @@ -262,12 +249,10 @@ func (h *Hub) handleRemoteEvent(evt *remoteEvent) { } } -// Subscribe enqueues a subscribe request for the given connection and channel. func (h *Hub) Subscribe(conn *Conn, channelID string) { h.subscribeCh <- &subscribeRequest{conn: conn, channelID: channelID} } -// disconnect sends a connection to the disconnect channel. func (h *Hub) disconnect(conn *Conn) { h.disconnectCh <- conn } diff --git a/go/internal/pusher/redis_bridge.go b/go/internal/pusher/redis_bridge.go index 2212f94..403b15f 100644 --- a/go/internal/pusher/redis_bridge.go +++ b/go/internal/pusher/redis_bridge.go @@ -23,22 +23,22 @@ const ( pubsubPrefix = "pusher:events:" ) -// redisEvent is published/received via Redis Pub/Sub for cross-pod communication. +// Wire format for cross-pod Pub/Sub. type redisEvent struct { Type string `json:"type"` // "join", "leave", "message" HumanID string `json:"humanId,omitempty"` // who triggered the event PodID string `json:"podId,omitempty"` // originating pod - Payload json.RawMessage `json:"payload,omitempty"` // for message events + Payload json.RawMessage `json:"payload,omitempty"` // message events only } -// RedisBridge handles cross-pod coordination via Redis Pub/Sub and presence tracking. +// RedisBridge handles cross-pod coordination via Redis Pub/Sub and presence +// tracking. type RedisBridge struct { client *redis.Client podID string - hub *Hub // set after hub is created + hub *Hub // wired post-construction; see SetHub } -// NewRedisBridge creates a new Redis bridge for cross-pod coordination. func NewRedisBridge(client *redis.Client, podID string) *RedisBridge { return &RedisBridge{ client: client, @@ -46,20 +46,20 @@ func NewRedisBridge(client *redis.Client, podID string) *RedisBridge { } } -// SetHub sets the hub reference. Called during initialization. +// SetHub resolves the circular dependency between Hub and RedisBridge. func (rb *RedisBridge) SetHub(hub *Hub) { rb.hub = hub } -// --- Presence management (called by hub goroutine) --- +// --- Presence management --- -// Subscribe adds a connection to a channel in Redis. -// Returns the current presence set for the channel. +// Subscribe records the connection in Redis and returns the channel's +// current deduplicated presence set. func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID string) ([]string, error) { key := channelConnsKey(channelID) field := rb.connField(connID) - // Check if humanID was already present before adding + // Snapshot before the insert so multi-tab joins don't double-emit. existingMembers, err := rb.client.HVals(ctx, key).Result() if err != nil && err != redis.Nil { return nil, fmt.Errorf("failed to get channel members: %w", err) @@ -67,12 +67,10 @@ func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID wasPresent := containsString(existingMembers, humanID) - // Add this connection if err := rb.client.HSet(ctx, key, field, humanID).Err(); err != nil { return nil, fmt.Errorf("failed to add connection to channel: %w", err) } - // Publish join event if this is a new humanID in the channel if !wasPresent { rb.publishEvent(ctx, channelID, redisEvent{ Type: TypeJoin, @@ -81,7 +79,6 @@ func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID }) } - // Return deduplicated presence set allMembers, err := rb.client.HVals(ctx, key).Result() if err != nil { return nil, fmt.Errorf("failed to get channel members: %w", err) @@ -89,7 +86,6 @@ func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID return deduplicateStrings(allMembers), nil } -// Unsubscribe removes a connection from a channel in Redis. func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, humanID string) error { key := channelConnsKey(channelID) field := rb.connField(connID) @@ -98,7 +94,7 @@ func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, human return fmt.Errorf("failed to remove connection from channel: %w", err) } - // Check if this humanID is still present via other connections + // Only emit leave once this humanID has no tabs left in the channel. remainingMembers, err := rb.client.HVals(ctx, key).Result() if err != nil && err != redis.Nil { return fmt.Errorf("failed to get remaining members: %w", err) @@ -112,7 +108,6 @@ func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, human }) } - // Clean up empty channel hash if len(remainingMembers) == 0 { rb.client.Del(ctx, key) } @@ -120,7 +115,6 @@ func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, human return nil } -// Broadcast publishes a message event to all pods. func (rb *RedisBridge) Broadcast(ctx context.Context, channelID, humanID string, payload json.RawMessage) { rb.publishEvent(ctx, channelID, redisEvent{ Type: TypeMessage, @@ -130,7 +124,6 @@ func (rb *RedisBridge) Broadcast(ctx context.Context, channelID, humanID string, }) } -// GetPresence returns the deduplicated humanIDs for the given channels. func (rb *RedisBridge) GetPresence(ctx context.Context, channelIDs []string) (map[string][]string, error) { result := make(map[string][]string, len(channelIDs)) for _, chID := range channelIDs { @@ -143,8 +136,7 @@ func (rb *RedisBridge) GetPresence(ctx context.Context, channelIDs []string) (ma return result, nil } -// GetAllConnectedHumanIDs scans all channel connection hashes in Redis and returns -// the deduplicated set of all humanIDs that have at least one active connection. +// Returns every humanID with at least one active connection cluster-wide. func (rb *RedisBridge) GetAllConnectedHumanIDs(ctx context.Context) ([]string, error) { allHumanIDs := make(map[string]bool) var cursor uint64 @@ -178,10 +170,9 @@ func (rb *RedisBridge) GetAllConnectedHumanIDs(ctx context.Context) ([]string, e return result, nil } -// --- Pub/Sub listener (runs in its own goroutine) --- +// --- Pub/Sub listener --- -// Listen subscribes to Redis Pub/Sub and forwards events to the local hub. -// Blocks until the context is cancelled. +// Listen forwards Redis Pub/Sub events to the local hub; blocks until ctx is cancelled. func (rb *RedisBridge) Listen(ctx context.Context) { pubsub := rb.client.PSubscribe(ctx, pubsubPrefix+"*") defer pubsub.Close() @@ -201,7 +192,7 @@ func (rb *RedisBridge) Listen(ctx context.Context) { } func (rb *RedisBridge) handlePubSubMessage(msg *redis.Message) { - // Extract channel ID from topic: "pusher:events:{channelID}" + // Topic: "pusher:events:{channelID}". channelID := strings.TrimPrefix(msg.Channel, pubsubPrefix) if channelID == "" { return @@ -213,7 +204,7 @@ func (rb *RedisBridge) handlePubSubMessage(msg *redis.Message) { return } - // Skip events originating from this pod — the local hub already handled them + // Same-pod events were already handled by the local hub. if event.PodID == rb.podID { return } @@ -222,20 +213,18 @@ func (rb *RedisBridge) handlePubSubMessage(msg *redis.Message) { return } - // Forward to local hub for delivery to local WebSocket connections rb.hub.remoteEventCh <- &remoteEvent{ channelID: channelID, event: event, } } -// --- Heartbeat + cleanup (runs in its own goroutine) --- +// --- Heartbeat + cleanup --- -// Heartbeat maintains this pod's liveness key and cleans up stale pods. +// Heartbeat refreshes this pod's liveness key and reaps stale pods on a tick. func (rb *RedisBridge) Heartbeat(ctx context.Context) { podKey := podKeyPrefix + rb.podID - // Initial heartbeat rb.client.Set(ctx, podKey, "alive", podHeartbeatTTL) heartbeatTicker := time.NewTicker(podHeartbeatInterval) @@ -246,7 +235,7 @@ func (rb *RedisBridge) Heartbeat(ctx context.Context) { for { select { case <-ctx.Done(): - // On shutdown, remove our pod key and clean up our connections + // On shutdown, drop our pod key and reclaim our connection slots. rb.client.Del(context.Background(), podKey) rb.cleanupPod(context.Background(), rb.podID) return @@ -259,7 +248,8 @@ func (rb *RedisBridge) Heartbeat(ctx context.Context) { } func (rb *RedisBridge) cleanupStalePods(ctx context.Context) { - // Scan all channel conn hashes for pod IDs, then check if those pods are still alive + // Collect every pod referenced in channel-conn hashes, then drop those + // whose liveness key has expired. var cursor uint64 knownPods := make(map[string]bool) alivePods := make(map[string]bool) @@ -290,7 +280,6 @@ func (rb *RedisBridge) cleanupStalePods(ctx context.Context) { } } - // Check which pods are still alive for podID := range knownPods { exists, err := rb.client.Exists(ctx, podKeyPrefix+podID).Result() if err != nil { @@ -301,7 +290,6 @@ func (rb *RedisBridge) cleanupStalePods(ctx context.Context) { } } - // Clean up dead pods for podID := range knownPods { if !alivePods[podID] { slog.Info("cleaning up stale pod", "podId", podID) @@ -328,7 +316,6 @@ func (rb *RedisBridge) cleanupPod(ctx context.Context, podID string) { for field, humanID := range fields { if extractPodID(field) == podID { rb.client.HDel(ctx, key, field) - // Check if this humanID is now gone from the channel remaining, _ := rb.client.HVals(ctx, key).Result() if !containsString(remaining, humanID) { rb.publishEvent(ctx, channelID, redisEvent{ @@ -369,15 +356,15 @@ func channelConnsKey(channelID string) string { return channelConnsPrefix + channelID + channelConnsSuffix } +// "pusher:ch:{channelID}:conns" → channelID func extractChannelID(redisKey string) string { - // "pusher:ch:{channelID}:conns" → channelID s := strings.TrimPrefix(redisKey, channelConnsPrefix) s = strings.TrimSuffix(s, channelConnsSuffix) return s } +// "{podID}:{connID}" → podID func extractPodID(field string) string { - // "{podID}:{connID}" → podID parts := strings.SplitN(field, ":", 2) if len(parts) == 2 { return parts[0] diff --git a/go/internal/pusher/server.go b/go/internal/pusher/server.go index 53349f3..cc32816 100644 --- a/go/internal/pusher/server.go +++ b/go/internal/pusher/server.go @@ -15,14 +15,12 @@ import ( type Server struct { pbpusher.UnimplementedPusherServiceServer - ctx context.Context // server-scoped context for graceful shutdown + ctx context.Context // server-scoped; cancelling closes all WebSockets gracefully hub *Hub bridge *RedisBridge authSvc auth.SessionReader } -// NewServer creates a new pusher server. The ctx controls the lifetime of all -// WebSocket connections — when cancelled, all connections are closed gracefully. func NewServer(ctx context.Context, hub *Hub, bridge *RedisBridge, authSvc auth.SessionReader) *Server { return &Server{ ctx: ctx, @@ -32,9 +30,8 @@ func NewServer(ctx context.Context, hub *Hub, bridge *RedisBridge, authSvc auth. } } -// HandleWebSocket handles the WebSocket upgrade and connection lifecycle. func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { - // Authenticate via query param (WebSocket upgrade can't use custom headers) + // Token rides in the query string — WebSocket upgrades can't carry custom headers. token := r.URL.Query().Get("token") if token == "" { http.Error(w, "token required", http.StatusUnauthorized) @@ -47,9 +44,8 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { return } - // Accept WebSocket upgrade ws, err := websocket.Accept(w, r, &websocket.AcceptOptions{ - // Allow all origins for now — CORS is handled at the gateway level + // CORS is enforced at the gateway. InsecureSkipVerify: true, }) if err != nil { @@ -62,25 +58,21 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { slog.Info("websocket connected", "connId", connID, "humanId", session.HumanId) - // Use server context, NOT r.Context(). After WebSocket upgrade, the HTTP - // request context can be cancelled by load balancers or Go's HTTP server, - // and nhooyr/websocket permanently closes the conn on any context error. + // Use the server context, not r.Context(): after upgrade the HTTP request + // context can be cancelled by load balancers and nhooyr/websocket would + // then permanently close the conn. ctx, cancel := context.WithCancel(s.ctx) defer cancel() - // Auto-subscribe to presence channel so this user appears online + // Auto-subscribe to the presence channel so this user appears online. s.hub.Subscribe(conn, "_presence:"+session.HumanId) - // WritePump runs in a separate goroutine go conn.WritePump(ctx) - - // ReadPump blocks until the connection closes conn.ReadPump(ctx, s.hub) slog.Info("websocket disconnected", "connId", connID, "humanId", session.HumanId) } -// GetOnlineHumanIds returns all currently connected human IDs. func (s *Server) GetOnlineHumanIds(ctx context.Context, _ *pbpusher.GetOnlineHumanIdsRequest) (*pbpusher.GetOnlineHumanIdsResponse, error) { humanIDs, err := s.bridge.GetAllConnectedHumanIDs(ctx) if err != nil { @@ -89,7 +81,6 @@ func (s *Server) GetOnlineHumanIds(ctx context.Context, _ *pbpusher.GetOnlineHum return &pbpusher.GetOnlineHumanIdsResponse{HumanIds: humanIDs}, nil } -// IsOnline checks whether specific humans are currently online. func (s *Server) IsOnline(ctx context.Context, req *pbpusher.IsOnlineRequest) (*pbpusher.IsOnlineResponse, error) { allOnline, err := s.bridge.GetAllConnectedHumanIDs(ctx) if err != nil { @@ -106,7 +97,6 @@ func (s *Server) IsOnline(ctx context.Context, req *pbpusher.IsOnlineRequest) (* return &pbpusher.IsOnlineResponse{Online: result}, nil } -// GetChannelPresence returns presence (human IDs) for specific channels. func (s *Server) GetChannelPresence(ctx context.Context, req *pbpusher.GetChannelPresenceRequest) (*pbpusher.GetChannelPresenceResponse, error) { presence, err := s.bridge.GetPresence(ctx, req.ChannelIds) if err != nil { diff --git a/go/internal/pusher/types.go b/go/internal/pusher/types.go index a518046..3c92990 100644 --- a/go/internal/pusher/types.go +++ b/go/internal/pusher/types.go @@ -18,14 +18,12 @@ const ( TypeError = "error" ) -// ClientMessage is a message sent from a WebSocket client to the server. type ClientMessage struct { Type string `json:"type"` Channel string `json:"channel,omitempty"` Payload json.RawMessage `json:"payload,omitempty"` } -// ServerMessage is a message sent from the server to a WebSocket client. type ServerMessage struct { Type string `json:"type"` Channel string `json:"channel,omitempty"` diff --git a/go/internal/redis.go b/go/internal/redis.go index 473f389..2fb052f 100644 --- a/go/internal/redis.go +++ b/go/internal/redis.go @@ -19,7 +19,7 @@ func ConnectAndTestRedis(db int) *redis.Client { redisAddr := fmt.Sprintf("%s:%s", redisHost, "6379") rdb := redis.NewClient(&redis.Options{ Addr: redisAddr, - Password: "", // no password set + Password: "", DB: db, }) diff --git a/go/internal/utils/env.go b/go/internal/utils/env.go index 64dafb7..125c241 100644 --- a/go/internal/utils/env.go +++ b/go/internal/utils/env.go @@ -6,13 +6,12 @@ import ( "github.com/sirupsen/logrus" ) -// enum of environment variables +// EnvVar enumerates the env vars referenced via this package. type EnvVar string const () -// MustGetEnv returns the value of the environment variable with the given key. -// panics if the variable is not set. +// MustGetEnv panics if the variable is unset. func MustGetEnv[T string | EnvVar](key T) string { keyString := string(key) value := os.Getenv(keyString) @@ -24,8 +23,7 @@ func MustGetEnv[T string | EnvVar](key T) string { return value } -// GetEnv returns the value of the environment variable with the given key. -// returns an empty string if the variable is not set. +// GetEnv returns "" if the variable is unset (and logs a warning). func GetEnv(key string) string { value := os.Getenv(key) if value == "" { diff --git a/go/internal/utils/optionals.go b/go/internal/utils/optionals.go index 84b58e8..f75979f 100644 --- a/go/internal/utils/optionals.go +++ b/go/internal/utils/optionals.go @@ -21,7 +21,6 @@ func CreateOptionalBool(input bool) *bool { return &input } -// OptionalString converts a non-nil *string to the respective string or returns "". func OptionalString(input *string) string { if input == nil { return "" @@ -30,7 +29,6 @@ func OptionalString(input *string) string { return *input } -// OptionalInt converts a non-nil *int to the respective int, otherwise returns 0. func OptionalInt(input *int) int { if input == nil { return 0 @@ -39,8 +37,7 @@ func OptionalInt(input *int) int { return *input } -// CreateOptionalInt when given a zero value int (0), it returns a nil *int. -// Otherwise, it gives a proper *int with valid value. +// Zero values become nil; the inverse of OptionalInt. func CreateOptionalInt(input int) *int { if input == 0 { return nil @@ -49,8 +46,7 @@ func CreateOptionalInt(input int) *int { return &input } -// CreateOptionalString when given an empty string, it returns a nil *string. -// Otherwise, it gives a proper *string with valid value. +// Empty string becomes nil; the inverse of OptionalString. func CreateOptionalString(input string) *string { if input == "" { return nil @@ -59,8 +55,7 @@ func CreateOptionalString(input string) *string { return &input } -// GetNumberFromString converts a string to a number. -// Returns error if the query is not a number. +// Returns an error if input contains non-digit characters or parses to <= 0. func GetNumberFromString(input string) (int, error) { for _, c := range input { if c < '0' || c > '9' { @@ -80,7 +75,6 @@ type Number interface { int | int8 | int16 | int32 | int64 | uint | uint8 | uint16 | uint32 | uint64 | float32 | float64 } -// OptionalNumber converts a non-nil *NUMBER to the respective number value or returns 0. func OptionalNumber[T Number](input *T) T { if input == nil { return 0 @@ -89,7 +83,7 @@ func OptionalNumber[T Number](input *T) T { return *input } -// CreateOptionalNumber when given an zero value NUMBER (0), it returns a nil *NUMBER, otherwise, it gives a proper *NUMBER with valid value. +// Zero values become nil; the inverse of OptionalNumber. func CreateOptionalNumber[T Number](input T) *T { if input == 0 { return nil diff --git a/go/internal/utils/random_generator.go b/go/internal/utils/random_generator.go index 022f091..6def086 100644 --- a/go/internal/utils/random_generator.go +++ b/go/internal/utils/random_generator.go @@ -10,7 +10,6 @@ const ( charsetNumbers = "0123456789" ) -// RandomString generates a random string of length n based on self defined charset func RandomString(length int) string { sb := strings.Builder{} sb.Grow(length) @@ -20,7 +19,6 @@ func RandomString(length int) string { return sb.String() } -// RandomStringNumbers func RandomStringNumbers(length int) string { sb := strings.Builder{} sb.Grow(length) diff --git a/go/internal/waitlist/service.go b/go/internal/waitlist/service.go index fddce24..4711079 100644 --- a/go/internal/waitlist/service.go +++ b/go/internal/waitlist/service.go @@ -34,11 +34,10 @@ const ( ) type Service interface { - // returns AlreadyInWaitlistError if already in the waitlist - // any other error is a failure + // AddToWaitlist returns AlreadyInWaitlistError if the email is already present. AddToWaitlist(ctx context.Context, email string, metadata map[string]string) error GetWaitlist(ctx context.Context, filter GetWaitlistFilter) ([]*WaitlistEntry, error) - // returns error if not found + // GetWaitlistEntryByEmail returns EntryNotFoundError if missing. GetWaitlistEntryByEmail(ctx context.Context, email string) (*WaitlistEntry, error) MarkWaitlistEntryInvited(ctx context.Context, email string) error }