FlowyLabs/llink
1
0
Fork 0
Code Issues 71 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: agentic comment cleanup

Browse Source
This commit is contained in:
talksik
2026-05-18 10:52:37 -07:00
parent 6f476fe773
commit 4421b8e832
45 changed files with 269 additions and 545 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/cmd/emailnotifierjob/main.go
+6 -31
View File
@@ -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).
go/cmd/memberreconciler/main.go
+7 -15
View File
@@ -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
go/cmd/orion/main.go
+1 -2
View File
@@ -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)
go/cmd/particleprocessorworker/main.go
+21 -35
View File
@@ -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:
go/cmd/pusherservice/main.go
+7 -19
View File
@@ -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())
go/cmd/transcodebackfill/main.go
+4 -12
View File
@@ -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++
go/internal/auth/reader.go
+2 -3
View File
@@ -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}
}
go/internal/auth/service.go
+5 -6
View File
@@ -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
go/internal/billing/service.go
+2 -3
View File
@@ -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),
go/internal/billing/usage_models.go
+1 -3
View File
@@ -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"`
go/internal/db/redis.go
+3 -3
View File
@@ -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
)
go/internal/depot/models.go
+3 -9
View File
@@ -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
go/internal/depot/service.go
+10 -18
View File
@@ -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
go/internal/handler/billing.go
+4 -8
View File
@@ -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 {
go/internal/handler/handler.go
+20 -41
View File
@@ -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
}
go/internal/handler/push_token.go
+3 -5
View File
@@ -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 {
go/internal/human/pushnotify/expo.go
+10 -12
View File
@@ -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
go/internal/human/pushnotify/notifier.go
+15 -19
View File
@@ -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))
go/internal/human/pushnotify/service.go
+8 -12
View File
@@ -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
}
go/internal/human/pushnotify/token.go
+6 -6
View File
@@ -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 (
go/internal/human/service.go
+2 -5
View File
@@ -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
}
go/internal/livekit/client.go
+2 -4
View File
@@ -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
}
go/internal/livestore/membershippublisher.go
+3 -4
View File
@@ -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
go/internal/media/helpers.go
+6 -10
View File
@@ -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",
go/internal/middleware/auth.go
+2 -9
View File
@@ -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 == "" {
go/internal/middleware/cors.go
+2 -3
View File
@@ -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
go/internal/network/reader.go
+3 -6
View File
@@ -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,
go/internal/network/repository.go
+3 -7
View File
@@ -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)
go/internal/network/service.go
+4 -7
View File
@@ -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 {
go/internal/particle/models.go
+4 -23
View File
@@ -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
go/internal/particle/networkmembershipchecker.go
-1
View File
@@ -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)
}
go/internal/particle/repository.go
+1 -1
View File
@@ -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)
go/internal/particle/service.go
+39 -82
View File
@@ -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
}
go/internal/pusher/authorizer.go
+7 -10
View File
@@ -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 {
go/internal/pusher/channel.go
+3 -5
View File
@@ -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 {
go/internal/pusher/conn.go
+4 -9
View File
@@ -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")
go/internal/pusher/hub.go
+6 -21
View File
@@ -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
}
go/internal/pusher/redis_bridge.go
+23 -36
View File
@@ -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]
go/internal/pusher/server.go
+7 -17
View File
@@ -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 {
go/internal/pusher/types.go
-2
View File
@@ -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"`
go/internal/redis.go
+1 -1
View File
@@ -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,
})
go/internal/utils/env.go
+3 -5
View File
@@ -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 == "" {
go/internal/utils/optionals.go
+4 -10
View File
@@ -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
go/internal/utils/random_generator.go
-2
View File
@@ -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)
go/internal/waitlist/service.go
+2 -3
View File
@@ -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
}