diff --git a/go/Makefile b/go/Makefile index df45e5b..f7403a5 100644 --- a/go/Makefile +++ b/go/Makefile @@ -38,7 +38,7 @@ PROD_REPO := us-west2-docker.pkg.dev/flowy-prod-440017/deployments .PHONY: migrate-dev migrate-dev: kubectl delete job migrations --ignore-not-found --context=dev - SKAFFOLD_DEFAULT_REPO=$(DEV_REPO) skaffold run -p migrations --kube-context dev + SKAFFOLD_DEFAULT_REPO=$(DEV_REPO) skaffold run -p migrations --kube-context dev --tail .PHONY: migrate-prod migrate-prod: diff --git a/go/cmd/emailnotifierjob/main.go b/go/cmd/emailnotifierjob/main.go index e536279..84871d9 100644 --- a/go/cmd/emailnotifierjob/main.go +++ b/go/cmd/emailnotifierjob/main.go @@ -11,6 +11,7 @@ import ( "cloud.google.com/go/firestore" pbaero "github.com/flowy-live/llink/genproto/aero" pbpusher "github.com/flowy-live/llink/genproto/llink/pusher" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/human" "github.com/flowy-live/llink/internal/network" @@ -67,7 +68,7 @@ func main() { // Initialize services humanSvc := human.NewService(db.Pool()) - networkSvc := network.NewService(db.Pool(), aeroSvc) + networkSvc := network.NewService(db.Pool(), aeroSvc, billing.Noop()) slog.Info("starting email notification cycle") if err := runNotificationCycle(ctx, firestoreClient, aeroSvc, pusherSvc, humanSvc, networkSvc); err != nil { diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index 9880ad8..b45faf0 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -12,6 +12,7 @@ import ( pbaero "github.com/flowy-live/llink/genproto/aero" "github.com/flowy-live/llink/internal" "github.com/flowy-live/llink/internal/auth" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/depot" "github.com/flowy-live/llink/internal/handler" @@ -35,14 +36,11 @@ func main() { port := utils.MustGetEnv("PORT") gcsBucket := utils.MustGetEnv("GCS_BUCKET") - // Initialize database db.Init() defer db.Cleanup() - // Initialize Redis for auth redisClient := redisForAuth() - // Initialize GCS client ctx := context.Background() storageClient, err := storage.NewClient(ctx) if err != nil { @@ -64,10 +62,23 @@ func main() { defer aeroServer.Close() aeroSvc := pbaero.NewPrimaryClient(aeroServer) - // Initialize services authSvc := auth.NewAuthService(redisClient, aeroSvc) humanSvc := human.NewService(db.Pool()) - networkSvc := network.NewService(db.Pool(), aeroSvc) + + billingSvc, err := billing.NewService(ctx, db.Pool(), billing.Config{ + SecretKey: utils.MustGetEnv("STRIPE_SECRET_KEY"), + WebhookSecret: utils.MustGetEnv("STRIPE_WEBHOOK_SECRET"), + PriceMonthlyID: utils.MustGetEnv("STRIPE_PRICE_PRO_MONTHLY"), + PriceAnnualID: utils.MustGetEnv("STRIPE_PRICE_PRO_ANNUAL"), + SuccessURL: utils.MustGetEnv("BILLING_SUCCESS_URL"), + CancelURL: utils.MustGetEnv("BILLING_CANCEL_URL"), + }) + if err != nil { + slog.Error("failed to initialize billing service", "error", err) + os.Exit(1) + } + + networkSvc := network.NewService(db.Pool(), aeroSvc, billingSvc) particleSvc := particle.NewService(db.Pool(), networkSvc) depotSvc := depot.NewService(db.Pool(), storageClient, depot.Config{ GoogleServiceAccountEmail: utils.MustGetEnv("GOOGLE_SERVICE_ACCOUNT_EMAIL"), @@ -76,7 +87,6 @@ func main() { waitlistSvc := waitlist.NewService(db.Pool(), aeroSvc) livekitClient := livekit.NewClient() - // Initialize Firestore client (for webhook-driven updates) gcpProject := utils.MustGetEnv("GCP_PROJECT") firestoreClient, err := firestore.NewClient(ctx, gcpProject) if err != nil { @@ -85,10 +95,8 @@ func main() { } defer firestoreClient.Close() - // Initialize handler - h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc, waitlistSvc, livekitClient, firestoreClient) + h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc, waitlistSvc, billingSvc, livekitClient, firestoreClient) - // Helper to wrap handlers with auth middleware withAuth := func(hf http.HandlerFunc) http.Handler { return middleware.Auth(authSvc)(http.HandlerFunc(hf)) } @@ -108,6 +116,7 @@ func main() { mux.HandleFunc("POST /auth/sign-in", h.SignIn) mux.HandleFunc("POST /waitlist", h.AddToWaitlist) mux.HandleFunc("POST /livekit/webhook", h.HandleLivekitWebhook) + mux.HandleFunc("POST /webhooks/stripe", h.HandleStripeWebhook) // ========================================================================== // Protected routes (auth required) @@ -127,6 +136,14 @@ func main() { mux.Handle("POST /networks/{id}/members", withAuth(h.AddMembersToNetwork)) // mux.Handle("DELETE /networks/{id}/members/{humanId}", withAuth(h.RemoveMemberFromNetwork)) + // Billing (network admin only; admin check happens inside each handler) + mux.Handle("GET /networks/{id}/billing", withAuth(h.GetNetworkBilling)) + mux.Handle("POST /networks/{id}/billing/checkout-session", withAuth(h.CreateCheckoutSession)) + mux.Handle("POST /networks/{id}/billing/portal-session", withAuth(h.CreatePortalSession)) + + // Freemium usage (any network member) + mux.Handle("GET /networks/{id}/usage", withAuth(h.GetNetworkUsage)) + // Network Invitations mux.Handle("GET /networks/{id}/invitations", withAuth(h.ListInvitationsForNetwork)) mux.Handle("DELETE /networks/{id}/invitations", withAuth(h.RevokeInvitation)) diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index 72528b5..6642d42 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -6,8 +6,10 @@ import ( "log" "log/slog" "os" + "strings" "time" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/depot" "github.com/flowy-live/llink/internal/particle" @@ -43,6 +45,7 @@ func main() { db.Init() defer db.Cleanup() processingRepo := particle.NewProcessingRepository(db.Pool()) + billingSvc := billing.NewServiceForWorker(db.Pool()) storageClient, err := storage.NewClient(ctx) if err != nil { @@ -104,6 +107,7 @@ func main() { updateParentLastChildCreatedAt(ctx, change.Doc) transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) + recordFreemiumUsage(ctx, billingSvc, change.Doc) if err := processingRepo.MarkProcessed(ctx, particleID); err != nil { slog.Error("failed to mark particle as processed", "particleID", particleID, "error", err) @@ -192,6 +196,59 @@ 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. +func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *firestore.DocumentSnapshot) { + rawType, err := doc.DataAt("type") + if err != nil { + slog.Error("failed to read particle type", "error", err, "particleID", doc.Ref.ID) + return + } + typeStr, ok := rawType.(string) + if !ok { + slog.Error("particle type is not a string", "particleID", doc.Ref.ID, "type", rawType) + return + } + particleType, err := particle.ParseParticleType(typeStr) + if err != nil { + slog.Error("invalid particle type", "error", err, "particleID", doc.Ref.ID) + return + } + // Containers (stream/folder) don't count as "messages" for the daily cap. + if particleType == particle.TypeStream || particleType == particle.TypeFolder { + return + } + + networkID, err := networkIDFromParticlePath(doc.Ref.Path) + if err != nil { + slog.Error("failed to derive network id", "error", err, "path", doc.Ref.Path) + return + } + + if err := billingSvc.IncrementDailyUsage(ctx, networkID, doc.CreateTime); err != nil { + slog.Error("failed to increment daily usage", "error", err, "networkID", networkID, "particleID", doc.Ref.ID) + } +} + +// networkIDFromParticlePath extracts the network id from a Firestore particle +// document path. Particles live at `networks/{network_id}/children/.../children/{id}` +// at arbitrary nesting depth, so the network id is always the second segment +// of the full doc path (which itself is rooted under the Firestore db path: +// `projects/.../documents/networks/{network_id}/...`). +func networkIDFromParticlePath(path string) (string, error) { + // doc.Ref.Path is the full resource path; find the "networks" collection + // and return the next segment. + segments := strings.Split(path, "/") + for i, seg := range segments { + if seg == "networks" && i+1 < len(segments) { + return segments[i+1], nil + } + } + return "", fmt.Errorf("no networks segment in path: %s", path) +} + // 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). diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 2e36865..0dab5cb 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -12,6 +12,7 @@ import ( "github.com/flowy-live/llink/internal" "github.com/flowy-live/llink/internal/auth" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/network" "github.com/flowy-live/llink/internal/pusher" @@ -36,8 +37,8 @@ func main() { pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) // Services - authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession - networkSvc := network.NewService(db.Pool(), nil) // nil aeroSvc — pusher never calls InviteByEmail + authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession + networkSvc := network.NewService(db.Pool(), nil, billing.Noop()) // nil aeroSvc / noop billing — pusher never mutates membership // Pod identity (use hostname in k8s, which is the pod name) podID, err := os.Hostname() diff --git a/go/go.mod b/go/go.mod index 2ff2e45..e9326e0 100644 --- a/go/go.mod +++ b/go/go.mod @@ -14,12 +14,14 @@ require ( github.com/redis/go-redis/v9 v9.17.2 github.com/sirupsen/logrus v1.9.3 github.com/stretchr/testify v1.11.1 + github.com/stripe/stripe-go/v85 v85.0.1 github.com/testcontainers/testcontainers-go v0.40.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.40.0 go.jetify.com/typeid v1.3.0 go.uber.org/mock v0.6.0 google.golang.org/grpc v1.79.1 google.golang.org/protobuf v1.36.11 + nhooyr.io/websocket v1.8.17 ) require ( @@ -176,5 +178,4 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/klog/v2 v2.110.1 // indirect - nhooyr.io/websocket v1.8.17 // indirect ) diff --git a/go/go.sum b/go/go.sum index c849499..db9694f 100644 --- a/go/go.sum +++ b/go/go.sum @@ -335,6 +335,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stripe/stripe-go/v85 v85.0.1 h1:vlIo5VHrR9GkYneH5D9YGOPwNDRD6LW/THhtx9zNs6M= +github.com/stripe/stripe-go/v85 v85.0.1/go.mod h1:5P+HGFenpWgak27T5Is6JMsmDfUC1yJnjhhmquz7kXw= github.com/testcontainers/testcontainers-go v0.40.0 h1:pSdJYLOVgLE8YdUY2FHQ1Fxu+aMnb6JfVz1mxk7OeMU= github.com/testcontainers/testcontainers-go v0.40.0/go.mod h1:FSXV5KQtX2HAMlm7U3APNyLkkap35zNLxukw9oBi/MY= github.com/testcontainers/testcontainers-go/modules/postgres v0.40.0 h1:s2bIayFXlbDFexo96y+htn7FzuhpXLYJNnIuglNKqOk= diff --git a/go/internal/billing/config.go b/go/internal/billing/config.go new file mode 100644 index 0000000..c64c321 --- /dev/null +++ b/go/internal/billing/config.go @@ -0,0 +1,37 @@ +package billing + +import ( + "fmt" + "strings" +) + +type Config struct { + SecretKey string + WebhookSecret string + PriceMonthlyID string + PriceAnnualID string + + SuccessURL string + CancelURL string +} + +func (c Config) Validate() error { + required := map[string]string{ + "SecretKey": c.SecretKey, + "WebhookSecret": c.WebhookSecret, + "PriceMonthlyID": c.PriceMonthlyID, + "PriceAnnualID": c.PriceAnnualID, + "SuccessURL": c.SuccessURL, + "CancelURL": c.CancelURL, + } + var missing []string + for name, value := range required { + if value == "" { + missing = append(missing, name) + } + } + if len(missing) > 0 { + return fmt.Errorf("billing config missing: %s", strings.Join(missing, ", ")) + } + return nil +} diff --git a/go/internal/billing/models.go b/go/internal/billing/models.go new file mode 100644 index 0000000..9a0a4b6 --- /dev/null +++ b/go/internal/billing/models.go @@ -0,0 +1,57 @@ +package billing + +import "time" + +type Cadence string + +const ( + CadenceMonthly Cadence = "monthly" + CadenceAnnual Cadence = "annual" +) + +func (c Cadence) IsValid() bool { + return c == CadenceMonthly || c == CadenceAnnual +} + +type Plan string + +const ( + PlanFree Plan = "free" + PlanPro Plan = "pro" +) + +// Subscription is the persisted projection of a Stripe subscription, +// reconciled on every webhook event. +type Subscription struct { + ID string + NetworkID string + StripeCustomerID string + Status string + PriceID string + Cadence Cadence + Quantity int + CancelAtPeriodEnd bool + CurrentPeriodStart time.Time + CurrentPeriodEnd time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type Status struct { + Plan Plan `json:"plan"` + PlanStatus string `json:"plan_status"` + Cadence *Cadence `json:"cadence"` + Seats int `json:"seats"` // 0 on free plan; sub.Quantity on pro + CurrentPeriodEnd *time.Time `json:"current_period_end"` + CancelAtPeriodEnd bool `json:"cancel_at_period_end"` + PriceMonthlyCents int64 `json:"price_monthly_cents"` + PriceAnnualCents int64 `json:"price_annual_cents"` +} + +type CheckoutParams struct { + NetworkID string + AdminHumanID string + AdminEmail string + Cadence Cadence + Seats int +} diff --git a/go/internal/billing/noop.go b/go/internal/billing/noop.go new file mode 100644 index 0000000..2eabafc --- /dev/null +++ b/go/internal/billing/noop.go @@ -0,0 +1,33 @@ +package billing + +import ( + "context" + "errors" + "time" +) + +// Noop returns a billing service for binaries that depend on network.Service +// but never mutate membership (jobs, pusher). Stripe isn't configured. +func Noop() Service { return noopService{} } + +type noopService struct{} + +var errNoopBilling = errors.New("billing: not configured in this process") + +func (noopService) GetStatus(context.Context, string) (*Status, error) { + return nil, errNoopBilling +} +func (noopService) CreateCheckoutSession(context.Context, CheckoutParams) (string, error) { + return "", errNoopBilling +} +func (noopService) CreatePortalSession(context.Context, string) (string, error) { + return "", errNoopBilling +} +func (noopService) SyncSeats(context.Context, string, int) error { return nil } +func (noopService) HandleWebhook(context.Context, []byte, string) error { return errNoopBilling } +func (noopService) GetUsage(context.Context, string) (*Usage, error) { + return nil, errNoopBilling +} +func (noopService) IncrementDailyUsage(context.Context, string, time.Time) error { + return errNoopBilling +} diff --git a/go/internal/billing/repository.go b/go/internal/billing/repository.go new file mode 100644 index 0000000..06ccf76 --- /dev/null +++ b/go/internal/billing/repository.go @@ -0,0 +1,112 @@ +package billing + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var errNotFound = errors.New("not found") + +type repository interface { + getSubscriptionByNetworkID(ctx context.Context, networkID string) (*Subscription, error) + upsertSubscription(ctx context.Context, sub *Subscription) error + deleteSubscriptionByID(ctx context.Context, subscriptionID string) error + + getStripeCustomerID(ctx context.Context, networkID string) (string, error) + setStripeCustomerID(ctx context.Context, networkID, customerID string) error +} + +type repositoryImpl struct { + pool *pgxpool.Pool +} + +func newRepository(pool *pgxpool.Pool) repository { + return &repositoryImpl{pool: pool} +} + +const subscriptionColumns = `id, network_id, stripe_customer_id, status, price_id, cadence, quantity, cancel_at_period_end, current_period_start, current_period_end, created_at, updated_at` + +func scanSubscription(row pgx.Row, s *Subscription) error { + return row.Scan( + &s.ID, &s.NetworkID, &s.StripeCustomerID, &s.Status, &s.PriceID, + &s.Cadence, &s.Quantity, &s.CancelAtPeriodEnd, + &s.CurrentPeriodStart, &s.CurrentPeriodEnd, + &s.CreatedAt, &s.UpdatedAt, + ) +} + +func (r *repositoryImpl) getSubscriptionByNetworkID(ctx context.Context, networkID string) (*Subscription, error) { + var s Subscription + err := scanSubscription( + r.pool.QueryRow(ctx, + `SELECT `+subscriptionColumns+` FROM network_subscriptions WHERE network_id = $1`, + networkID, + ), + &s, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errNotFound + } + if err != nil { + return nil, err + } + return &s, nil +} + +func (r *repositoryImpl) upsertSubscription(ctx context.Context, sub *Subscription) error { + _, err := r.pool.Exec(ctx, ` + INSERT INTO network_subscriptions ( + id, network_id, stripe_customer_id, status, price_id, cadence, + quantity, cancel_at_period_end, current_period_start, current_period_end + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + ON CONFLICT (id) DO UPDATE SET + stripe_customer_id = EXCLUDED.stripe_customer_id, + status = EXCLUDED.status, + price_id = EXCLUDED.price_id, + cadence = EXCLUDED.cadence, + quantity = EXCLUDED.quantity, + cancel_at_period_end = EXCLUDED.cancel_at_period_end, + current_period_start = EXCLUDED.current_period_start, + current_period_end = EXCLUDED.current_period_end, + updated_at = NOW() + `, + sub.ID, sub.NetworkID, sub.StripeCustomerID, sub.Status, sub.PriceID, sub.Cadence, + sub.Quantity, sub.CancelAtPeriodEnd, sub.CurrentPeriodStart, sub.CurrentPeriodEnd, + ) + return err +} + +func (r *repositoryImpl) deleteSubscriptionByID(ctx context.Context, subscriptionID string) error { + _, err := r.pool.Exec(ctx, + `DELETE FROM network_subscriptions WHERE id = $1`, + subscriptionID, + ) + return err +} + +func (r *repositoryImpl) getStripeCustomerID(ctx context.Context, networkID string) (string, error) { + var customerID string + err := r.pool.QueryRow(ctx, + `SELECT stripe_customer_id FROM network_stripe_customers WHERE network_id = $1`, + networkID, + ).Scan(&customerID) + if errors.Is(err, pgx.ErrNoRows) { + return "", errNotFound + } + if err != nil { + return "", err + } + return customerID, nil +} + +func (r *repositoryImpl) setStripeCustomerID(ctx context.Context, networkID, customerID string) error { + _, err := r.pool.Exec(ctx, ` + INSERT INTO network_stripe_customers (network_id, stripe_customer_id) + VALUES ($1, $2) + ON CONFLICT (network_id) DO NOTHING + `, networkID, customerID) + return err +} diff --git a/go/internal/billing/service.go b/go/internal/billing/service.go new file mode 100644 index 0000000..0e51a85 --- /dev/null +++ b/go/internal/billing/service.go @@ -0,0 +1,289 @@ +package billing + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stripe/stripe-go/v85" + billingportalsession "github.com/stripe/stripe-go/v85/billingportal/session" + checkoutsession "github.com/stripe/stripe-go/v85/checkout/session" + stripecustomer "github.com/stripe/stripe-go/v85/customer" + stripeprice "github.com/stripe/stripe-go/v85/price" + stripesub "github.com/stripe/stripe-go/v85/subscription" +) + +type Service interface { + GetStatus(ctx context.Context, networkID string) (*Status, error) + CreateCheckoutSession(ctx context.Context, p CheckoutParams) (url string, err error) + CreatePortalSession(ctx context.Context, networkID string) (url string, err error) + // SyncSeats updates the Stripe subscription quantity with proration. + // No-op if the network has no active subscription. + SyncSeats(ctx context.Context, networkID string, seats int) error + HandleWebhook(ctx context.Context, payload []byte, signature string) error + + // GetUsage reports today's freemium quota state for a network. + // Pro networks get Limit=nil (unlimited); free networks get Limit=&FreemiumDailyLimit. + GetUsage(ctx context.Context, networkID string) (*Usage, error) + IncrementDailyUsage(ctx context.Context, networkID string, at time.Time) error +} + +var ( + ErrNoActiveSubscription = errors.New("network has no stripe customer yet") + ErrInvalidCadence = errors.New("invalid billing cadence") +) + +type serviceImpl struct { + cfg Config + repo repository + usageRepo usageRepository + priceMonthlyCents int64 + priceAnnualCents int64 +} + +func NewService(ctx context.Context, pool *pgxpool.Pool, cfg Config) (Service, error) { + if err := cfg.Validate(); err != nil { + return nil, err + } + stripe.Key = cfg.SecretKey + + monthly, err := stripeprice.Get(cfg.PriceMonthlyID, &stripe.PriceParams{ + Params: stripe.Params{Context: ctx}, + }) + if err != nil { + return nil, fmt.Errorf("fetch stripe monthly price: %w", err) + } + annual, err := stripeprice.Get(cfg.PriceAnnualID, &stripe.PriceParams{ + Params: stripe.Params{Context: ctx}, + }) + if err != nil { + return nil, fmt.Errorf("fetch stripe annual price: %w", err) + } + + return &serviceImpl{ + cfg: cfg, + repo: newRepository(pool), + usageRepo: newUsageRepository(pool), + priceMonthlyCents: monthly.UnitAmount, + priceAnnualCents: annual.UnitAmount, + }, 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). +func NewServiceForWorker(pool *pgxpool.Pool) Service { + return &serviceImpl{ + usageRepo: newUsageRepository(pool), + repo: newRepository(pool), + } +} + +func (s *serviceImpl) GetStatus(ctx context.Context, networkID string) (*Status, error) { + sub, err := s.repo.getSubscriptionByNetworkID(ctx, networkID) + if err != nil && !errors.Is(err, errNotFound) { + return nil, err + } + + status := &Status{ + Plan: PlanFree, + PlanStatus: "active", + PriceMonthlyCents: s.priceMonthlyCents, + PriceAnnualCents: s.priceAnnualCents, + } + + if sub != nil { + cadence := sub.Cadence + periodEnd := sub.CurrentPeriodEnd + // past_due keeps access: Stripe still considers the subscription live + // during the dunning window. + switch stripe.SubscriptionStatus(sub.Status) { + case stripe.SubscriptionStatusActive, + stripe.SubscriptionStatusTrialing, + stripe.SubscriptionStatusPastDue: + status.Plan = PlanPro + } + status.PlanStatus = sub.Status + status.Cadence = &cadence + status.Seats = sub.Quantity + status.CurrentPeriodEnd = &periodEnd + status.CancelAtPeriodEnd = sub.CancelAtPeriodEnd + } + + return status, nil +} + +func (s *serviceImpl) CreateCheckoutSession(ctx context.Context, p CheckoutParams) (string, error) { + if !p.Cadence.IsValid() { + return "", ErrInvalidCadence + } + if p.Seats < 1 { + return "", fmt.Errorf("seats must be >= 1") + } + + customerID, err := s.ensureStripeCustomer(ctx, p) + if err != nil { + return "", fmt.Errorf("ensure stripe customer: %w", err) + } + + priceID := s.cfg.PriceAnnualID + switch p.Cadence { + case CadenceAnnual: + priceID = s.cfg.PriceAnnualID + break + case CadenceMonthly: + priceID = s.cfg.PriceMonthlyID + break + } + + params := &stripe.CheckoutSessionParams{ + Mode: stripe.String(string(stripe.CheckoutSessionModeSubscription)), + Customer: stripe.String(customerID), + ClientReferenceID: stripe.String(p.NetworkID), + SuccessURL: stripe.String(s.cfg.SuccessURL), + CancelURL: stripe.String(s.cfg.CancelURL), + LineItems: []*stripe.CheckoutSessionLineItemParams{{ + Price: stripe.String(priceID), + Quantity: stripe.Int64(int64(p.Seats)), + }}, + SubscriptionData: &stripe.CheckoutSessionSubscriptionDataParams{ + Metadata: map[string]string{ + "network_id": p.NetworkID, + "admin_human_id": p.AdminHumanID, + "cadence": string(p.Cadence), + }, + }, + } + params.Context = ctx + + sess, err := checkoutsession.New(params) + if err != nil { + return "", fmt.Errorf("stripe checkout: %w", err) + } + return sess.URL, nil +} + +func (s *serviceImpl) CreatePortalSession(ctx context.Context, networkID string) (string, error) { + sub, err := s.repo.getSubscriptionByNetworkID(ctx, networkID) + if errors.Is(err, errNotFound) { + return "", ErrNoActiveSubscription + } + if err != nil { + return "", err + } + + params := &stripe.BillingPortalSessionParams{ + Customer: stripe.String(sub.StripeCustomerID), + ReturnURL: stripe.String(s.cfg.SuccessURL), + } + params.Context = ctx + + sess, err := billingportalsession.New(params) + if err != nil { + return "", fmt.Errorf("stripe portal: %w", err) + } + return sess.URL, nil +} + +func (s *serviceImpl) SyncSeats(ctx context.Context, networkID string, seats int) error { + sub, err := s.repo.getSubscriptionByNetworkID(ctx, networkID) + if errors.Is(err, errNotFound) { + return nil + } + if err != nil { + return err + } + if sub.Quantity == seats { + return nil + } + + liveSub, err := stripesub.Get(sub.ID, &stripe.SubscriptionParams{ + Params: stripe.Params{Context: ctx}, + }) + if err != nil { + return fmt.Errorf("fetch stripe subscription: %w", err) + } + if len(liveSub.Items.Data) == 0 { + return fmt.Errorf("stripe subscription %s has no items", sub.ID) + } + + params := &stripe.SubscriptionParams{ + ProrationBehavior: stripe.String(string(stripe.SubscriptionSchedulePhaseProrationBehaviorCreateProrations)), + Items: []*stripe.SubscriptionItemsParams{{ + ID: stripe.String(liveSub.Items.Data[0].ID), + Quantity: stripe.Int64(int64(seats)), + }}, + } + params.Context = ctx + + if _, err := stripesub.Update(sub.ID, params); err != nil { + return fmt.Errorf("update stripe subscription quantity: %w", err) + } + // customer.subscription.updated webhook arrives within seconds and + // reconciles quantity in our DB. + return nil +} + +func (s *serviceImpl) IncrementDailyUsage(ctx context.Context, networkID string, at time.Time) error { + return s.usageRepo.incrementDaily(ctx, networkID, at) +} + +func (s *serviceImpl) GetUsage(ctx context.Context, networkID string) (*Usage, error) { + now := time.Now() + used, err := s.usageRepo.getDaily(ctx, networkID, now) + if err != nil { + return nil, fmt.Errorf("read daily usage: %w", err) + } + + sub, err := s.GetStatus(ctx, networkID) + if err != nil { + return nil, fmt.Errorf("unable to get network billing status: %w", err) + } + + u := &Usage{ + Plan: sub.Plan, + Used: used, + ResetAt: nextUTCMidnight(now), + Limit: nil, + } + if sub.Plan == PlanFree { + limit := FreemiumDailyLimit + u.Limit = &limit + } + return u, nil +} + +func nextUTCMidnight(now time.Time) time.Time { + utc := now.UTC() + return time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC).Add(24 * time.Hour) +} + +func (s *serviceImpl) ensureStripeCustomer(ctx context.Context, p CheckoutParams) (string, error) { + existing, err := s.repo.getStripeCustomerID(ctx, p.NetworkID) + if err == nil { + return existing, nil + } + if !errors.Is(err, errNotFound) { + return "", err + } + + params := &stripe.CustomerParams{ + Email: stripe.String(p.AdminEmail), + Metadata: map[string]string{ + "network_id": p.NetworkID, + "admin_human_id": p.AdminHumanID, + }, + } + params.Context = ctx + + cust, err := stripecustomer.New(params) + if err != nil { + return "", fmt.Errorf("create stripe customer: %w", err) + } + if err := s.repo.setStripeCustomerID(ctx, p.NetworkID, cust.ID); err != nil { + return "", fmt.Errorf("persist stripe customer id: %w", err) + } + return cust.ID, nil +} diff --git a/go/internal/billing/usage_models.go b/go/internal/billing/usage_models.go new file mode 100644 index 0000000..6e8ec00 --- /dev/null +++ b/go/internal/billing/usage_models.go @@ -0,0 +1,15 @@ +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. +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"` + Limit *int `json:"limit"` // nil = unlimited (pro) + ResetAt time.Time `json:"reset_at"` +} diff --git a/go/internal/billing/usage_repository.go b/go/internal/billing/usage_repository.go new file mode 100644 index 0000000..79364a4 --- /dev/null +++ b/go/internal/billing/usage_repository.go @@ -0,0 +1,49 @@ +package billing + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type usageRepository interface { + incrementDaily(ctx context.Context, networkID string, at time.Time) error + getDaily(ctx context.Context, networkID string, at time.Time) (int, error) +} + +type usageRepositoryImpl struct { + pool *pgxpool.Pool +} + +func newUsageRepository(pool *pgxpool.Pool) usageRepository { + return &usageRepositoryImpl{pool: pool} +} + +func (r *usageRepositoryImpl) incrementDaily(ctx context.Context, networkID string, at time.Time) error { + _, err := r.pool.Exec(ctx, ` + INSERT INTO network_message_usage (network_id, usage_date, message_count, updated_at) + VALUES ($1, ($2 AT TIME ZONE 'UTC')::date, 1, NOW()) + ON CONFLICT (network_id, usage_date) DO UPDATE + SET message_count = network_message_usage.message_count + 1, + updated_at = NOW() + `, networkID, at) + return err +} + +func (r *usageRepositoryImpl) getDaily(ctx context.Context, networkID string, at time.Time) (int, error) { + var count int + err := r.pool.QueryRow(ctx, ` + SELECT message_count FROM network_message_usage + WHERE network_id = $1 AND usage_date = ($2 AT TIME ZONE 'UTC')::date + `, networkID, at).Scan(&count) + if errors.Is(err, pgx.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, err + } + return count, nil +} diff --git a/go/internal/billing/webhooks.go b/go/internal/billing/webhooks.go new file mode 100644 index 0000000..9dab258 --- /dev/null +++ b/go/internal/billing/webhooks.go @@ -0,0 +1,115 @@ +package billing + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/stripe/stripe-go/v85" + "github.com/stripe/stripe-go/v85/webhook" +) + +func (s *serviceImpl) HandleWebhook(ctx context.Context, payload []byte, signature string) error { + event, err := webhook.ConstructEvent(payload, signature, s.cfg.WebhookSecret) + if err != nil { + return fmt.Errorf("verify stripe signature: %w", err) + } + + slog.Info("stripe webhook", "type", event.Type, "id", event.ID) + + switch event.Type { + case "checkout.session.completed": + // subscription.created fires right after with full detail; we handle + // the subscription there. + return nil + case "customer.subscription.created", "customer.subscription.updated": + return s.handleSubscriptionUpsert(ctx, event) + case "customer.subscription.deleted": + return s.handleSubscriptionDeleted(ctx, event) + default: + return nil + } +} + +func (s *serviceImpl) handleSubscriptionUpsert(ctx context.Context, event stripe.Event) error { + var sub stripe.Subscription + if err := json.Unmarshal(event.Data.Raw, &sub); err != nil { + return fmt.Errorf("decode subscription: %w", err) + } + + networkID := sub.Metadata["network_id"] + if networkID == "" { + return fmt.Errorf("subscription %s missing network_id metadata", sub.ID) + } + + local, err := subscriptionFromStripe(&sub, networkID) + if err != nil { + return err + } + if err := s.repo.upsertSubscription(ctx, local); err != nil { + return fmt.Errorf("upsert subscription: %w", err) + } + return nil +} + +func (s *serviceImpl) handleSubscriptionDeleted(ctx context.Context, event stripe.Event) error { + var sub stripe.Subscription + if err := json.Unmarshal(event.Data.Raw, &sub); err != nil { + return fmt.Errorf("decode subscription: %w", err) + } + + networkID := sub.Metadata["network_id"] + if networkID == "" { + return fmt.Errorf("subscription %s missing network_id metadata", sub.ID) + } + + if err := s.repo.deleteSubscriptionByID(ctx, sub.ID); err != nil { + return fmt.Errorf("delete subscription: %w", err) + } + return nil +} + +func subscriptionFromStripe(sub *stripe.Subscription, networkID string) (*Subscription, error) { + if len(sub.Items.Data) == 0 { + return nil, fmt.Errorf("subscription %s has no items", sub.ID) + } + item := sub.Items.Data[0] + cadence, err := cadenceFromInterval(item.Price) + if err != nil { + return nil, fmt.Errorf("subscription %s: %w", sub.ID, err) + } + + var customerID string + if sub.Customer != nil { + customerID = sub.Customer.ID + } + + return &Subscription{ + ID: sub.ID, + NetworkID: networkID, + StripeCustomerID: customerID, + Status: string(sub.Status), + PriceID: item.Price.ID, + Cadence: cadence, + Quantity: int(item.Quantity), + CancelAtPeriodEnd: sub.CancelAtPeriodEnd, + CurrentPeriodStart: time.Unix(item.CurrentPeriodStart, 0).UTC(), + CurrentPeriodEnd: time.Unix(item.CurrentPeriodEnd, 0).UTC(), + }, nil +} + +func cadenceFromInterval(price *stripe.Price) (Cadence, error) { + if price == nil || price.Recurring == nil { + return "", fmt.Errorf("price is not recurring") + } + switch price.Recurring.Interval { + case stripe.PriceRecurringIntervalMonth: + return CadenceMonthly, nil + case stripe.PriceRecurringIntervalYear: + return CadenceAnnual, nil + default: + return "", fmt.Errorf("unsupported price interval %q", price.Recurring.Interval) + } +} diff --git a/go/internal/handler/billing.go b/go/internal/handler/billing.go new file mode 100644 index 0000000..2d17206 --- /dev/null +++ b/go/internal/handler/billing.go @@ -0,0 +1,212 @@ +package handler + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + + "github.com/flowy-live/llink/internal/billing" + "github.com/flowy-live/llink/internal/middleware" + "github.com/flowy-live/llink/internal/network" +) + +type CreateCheckoutSessionRequest struct { + Cadence string `json:"cadence"` +} + +type CheckoutSessionResponse struct { + URL string `json:"url"` +} + +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. +func (h *Handler) GetNetworkUsage(w http.ResponseWriter, r *http.Request) { + humanId, ok := middleware.HumanIdFromContext(r.Context()) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + networkID := r.PathValue("id") + if networkID == "" { + http.Error(w, "network id is required", http.StatusBadRequest) + return + } + + isMember, err := h.networkSvc.IsMember(r.Context(), networkID, humanId) + if err != nil { + slog.Error("failed to check network membership", "error", err, "network_id", networkID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + if !isMember { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + + usage, err := h.billingSvc.GetUsage(r.Context(), networkID) + if err != nil { + slog.Error("failed to get network usage", "error", err, "network_id", networkID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + writeJSON(w, usage) +} + +func (h *Handler) GetNetworkBilling(w http.ResponseWriter, r *http.Request) { + net, _, ok := h.loadNetworkForAdmin(w, r) + if !ok { + return + } + + status, err := h.billingSvc.GetStatus(r.Context(), net.ID) + if err != nil { + slog.Error("failed to get billing status", "error", err, "network_id", net.ID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + writeJSON(w, status) +} + +func (h *Handler) CreateCheckoutSession(w http.ResponseWriter, r *http.Request) { + net, adminHumanId, ok := h.loadNetworkForAdmin(w, r) + if !ok { + return + } + + var req CreateCheckoutSessionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + + cadence := billing.Cadence(req.Cadence) + if !cadence.IsValid() { + http.Error(w, "invalid cadence", http.StatusBadRequest) + return + } + + adminHuman, err := h.humanSvc.GetByID(r.Context(), adminHumanId) + if err != nil { + slog.Error("failed to load admin human", "error", err, "human_id", adminHumanId) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + seats, err := h.networkSvc.CountSeats(r.Context(), net.ID) + if err != nil { + slog.Error("failed to count seats", "error", err, "network_id", net.ID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + url, err := h.billingSvc.CreateCheckoutSession(r.Context(), billing.CheckoutParams{ + NetworkID: net.ID, + AdminHumanID: adminHumanId, + AdminEmail: adminHuman.Email, + Cadence: cadence, + Seats: seats, + }) + if err != nil { + slog.Error("failed to create checkout session", "error", err, "network_id", net.ID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + writeJSON(w, CheckoutSessionResponse{URL: url}) +} + +func (h *Handler) CreatePortalSession(w http.ResponseWriter, r *http.Request) { + net, _, ok := h.loadNetworkForAdmin(w, r) + if !ok { + return + } + + url, err := h.billingSvc.CreatePortalSession(r.Context(), net.ID) + if errors.Is(err, billing.ErrNoActiveSubscription) { + http.Error(w, "no active subscription", http.StatusConflict) + return + } + if err != nil { + slog.Error("failed to create portal session", "error", err, "network_id", net.ID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + writeJSON(w, PortalSessionResponse{URL: url}) +} + +const maxStripeWebhookBytes = 1 << 20 // 1 MiB + +func (h *Handler) HandleStripeWebhook(w http.ResponseWriter, r *http.Request) { + payload, err := io.ReadAll(io.LimitReader(r.Body, maxStripeWebhookBytes)) + if err != nil { + slog.Warn("stripe webhook: failed to read body", "error", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + signature := r.Header.Get("Stripe-Signature") + + if err := h.billingSvc.HandleWebhook(r.Context(), payload, signature); err != nil { + slog.Error("stripe webhook failed", "error", err) + formattedErr := fmt.Errorf("webhook processing failed: %w", err) + http.Error(w, formattedErr.Error(), http.StatusBadRequest) + return + } + + 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. +func (h *Handler) loadNetworkForAdmin(w http.ResponseWriter, r *http.Request) (*network.Network, string, bool) { + humanId, ok := middleware.HumanIdFromContext(r.Context()) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return nil, "", false + } + + networkID := r.PathValue("id") + if networkID == "" { + http.Error(w, "network id is required", http.StatusBadRequest) + return nil, "", false + } + + net, err := h.networkSvc.GetByID(r.Context(), networkID) + if errors.Is(err, network.ErrNotFound) { + http.Error(w, "network not found", http.StatusNotFound) + return nil, "", false + } + if err != nil { + slog.Error("failed to load network", "error", err, "network_id", networkID) + http.Error(w, "internal server error", http.StatusInternalServerError) + return nil, "", false + } + + if net.AdminHumanId != humanId { + http.Error(w, "forbidden", http.StatusForbidden) + return nil, "", false + } + + return net, humanId, true +} + +func writeJSON(w http.ResponseWriter, body any) { + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(body); err != nil { + slog.Error("failed to write json", "error", err) + } +} diff --git a/go/internal/handler/handler.go b/go/internal/handler/handler.go index 3db4cee..d6d90e8 100644 --- a/go/internal/handler/handler.go +++ b/go/internal/handler/handler.go @@ -12,6 +12,7 @@ import ( "cloud.google.com/go/firestore" "github.com/flowy-live/llink/internal/auth" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/depot" "github.com/flowy-live/llink/internal/human" "github.com/flowy-live/llink/internal/livekit" @@ -30,11 +31,22 @@ type Handler struct { particleSvc particle.Service depotSvc depot.Service waitlistSvc waitlist.Service + billingSvc billing.Service livekitClient livekit.Client firestoreClient *firestore.Client } -func NewHandler(authSvc auth.AuthService, humanSvc human.Service, networkSvc network.Service, particleSvc particle.Service, depotSvc depot.Service, waitlistSvc waitlist.Service, livekitClient livekit.Client, firestoreClient *firestore.Client) *Handler { +func NewHandler( + authSvc auth.AuthService, + humanSvc human.Service, + networkSvc network.Service, + particleSvc particle.Service, + depotSvc depot.Service, + waitlistSvc waitlist.Service, + billingSvc billing.Service, + livekitClient livekit.Client, + firestoreClient *firestore.Client, +) *Handler { return &Handler{ authSvc: authSvc, humanSvc: humanSvc, @@ -42,6 +54,7 @@ func NewHandler(authSvc auth.AuthService, humanSvc human.Service, networkSvc net particleSvc: particleSvc, depotSvc: depotSvc, waitlistSvc: waitlistSvc, + billingSvc: billingSvc, livekitClient: livekitClient, firestoreClient: firestoreClient, } diff --git a/go/internal/network/models.go b/go/internal/network/models.go index a0906b1..9a3cc76 100644 --- a/go/internal/network/models.go +++ b/go/internal/network/models.go @@ -3,11 +3,11 @@ package network import "time" type Network struct { - ID string - Name string - AdminHumanId string - MemberHumanIds []string - CreatedAt time.Time + ID string + Name string + AdminHumanId string + MemberHumanIds []string + CreatedAt time.Time } type Invitation struct { diff --git a/go/internal/network/repository.go b/go/internal/network/repository.go index b7db2f7..523a89b 100644 --- a/go/internal/network/repository.go +++ b/go/internal/network/repository.go @@ -5,10 +5,20 @@ import ( "errors" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "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. +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) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + var errNotFound = errors.New("not found") type networkIDPrefix struct{} @@ -30,9 +40,10 @@ type repository interface { getByID(ctx context.Context, id string) (*Network, error) updateName(ctx context.Context, id, name string) error delete(ctx context.Context, id string) error - addMember(ctx context.Context, networkID, humanId string) error - removeMember(ctx context.Context, networkID, humanId string) error + addMember(ctx context.Context, db dbtx, networkID, humanId string) error + removeMember(ctx context.Context, db dbtx, networkID, humanId string) error getMemberHumanIds(ctx context.Context, networkID string) ([]string, error) + countSeats(ctx context.Context, db dbtx, networkID string) (int, error) getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error) isMember(ctx context.Context, networkID, humanId string) (bool, error) listAll(ctx context.Context) ([]*Network, error) @@ -41,7 +52,15 @@ type repository interface { createInvitation(ctx context.Context, networkID, email string) error getInvitationsByEmail(ctx context.Context, email string) ([]*Invitation, error) getInvitationsByNetwork(ctx context.Context, networkID string) ([]*Invitation, error) - deleteInvitation(ctx context.Context, networkID, email string) error + 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. +const networkColumns = `id, name, admin_human_id, created_at` + +func scanNetwork(row pgx.Row, n *Network) error { + return row.Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.CreatedAt) } type repositoryImpl struct { @@ -59,12 +78,12 @@ func (r *repositoryImpl) create(ctx context.Context, name, adminHumanId string) } var n Network - err = r.pool.QueryRow(ctx, + row := r.pool.QueryRow(ctx, `INSERT INTO networks (id, name, admin_human_id) VALUES ($1, $2, $3) - RETURNING id, name, admin_human_id, created_at`, + RETURNING `+networkColumns, id.String(), name, adminHumanId, - ).Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.CreatedAt) - if err != nil { + ) + if err := scanNetwork(row, &n); err != nil { return nil, err } @@ -74,21 +93,22 @@ func (r *repositoryImpl) create(ctx context.Context, name, adminHumanId string) func (r *repositoryImpl) getByID(ctx context.Context, id string) (*Network, error) { var n Network - err := r.pool.QueryRow(ctx, - `SELECT id, name, admin_human_id, created_at FROM networks WHERE id = $1`, + row := r.pool.QueryRow(ctx, + `SELECT `+networkColumns+` FROM networks WHERE id = $1`, id, - ).Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.CreatedAt) - if err != nil { + ) + if err := scanNetwork(row, &n); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, errNotFound } return nil, err } - n.MemberHumanIds, err = r.getMemberHumanIds(ctx, id) + memberIds, err := r.getMemberHumanIds(ctx, id) if err != nil { return nil, err } + n.MemberHumanIds = memberIds return &n, nil } @@ -118,8 +138,8 @@ func (r *repositoryImpl) delete(ctx context.Context, id string) error { return nil } -func (r *repositoryImpl) addMember(ctx context.Context, networkID, humanId string) error { - _, err := r.pool.Exec(ctx, +func (r *repositoryImpl) addMember(ctx context.Context, db dbtx, networkID, humanId string) error { + _, err := db.Exec(ctx, `INSERT INTO network_members (network_id, human_id) VALUES ($1, $2) ON CONFLICT (network_id, human_id) DO NOTHING`, networkID, humanId, @@ -127,14 +147,23 @@ func (r *repositoryImpl) addMember(ctx context.Context, networkID, humanId strin return err } -func (r *repositoryImpl) removeMember(ctx context.Context, networkID, humanId string) error { - _, err := r.pool.Exec(ctx, +func (r *repositoryImpl) removeMember(ctx context.Context, db dbtx, networkID, humanId string) error { + _, err := db.Exec(ctx, `DELETE FROM network_members WHERE network_id = $1 AND human_id = $2`, networkID, humanId, ) return err } +func (r *repositoryImpl) countSeats(ctx context.Context, db dbtx, networkID string) (int, error) { + var count int + err := db.QueryRow(ctx, + `SELECT COUNT(*) FROM network_members WHERE network_id = $1`, + networkID, + ).Scan(&count) + return count, err +} + func (r *repositoryImpl) getMemberHumanIds(ctx context.Context, networkID string) ([]string, error) { rows, err := r.pool.Query(ctx, `SELECT human_id FROM network_members WHERE network_id = $1`, @@ -158,10 +187,10 @@ func (r *repositoryImpl) getMemberHumanIds(ctx context.Context, networkID string func (r *repositoryImpl) getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error) { rows, err := r.pool.Query(ctx, - `SELECT n.id, n.name, n.admin_human_id, n.created_at - FROM networks n - WHERE n.admin_human_id = $1 - OR EXISTS (SELECT 1 FROM network_members nm WHERE nm.network_id = n.id AND nm.human_id = $1)`, + `SELECT `+networkColumns+` + FROM networks + WHERE admin_human_id = $1 + OR EXISTS (SELECT 1 FROM network_members nm WHERE nm.network_id = id AND nm.human_id = $1)`, humanId, ) if err != nil { @@ -172,7 +201,7 @@ func (r *repositoryImpl) getNetworksForHuman(ctx context.Context, humanId string var networks []*Network for rows.Next() { var n Network - if err := rows.Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.CreatedAt); err != nil { + if err := scanNetwork(rows, &n); err != nil { return nil, err } networks = append(networks, &n) @@ -205,7 +234,7 @@ func (r *repositoryImpl) isMember(ctx context.Context, networkID, humanId string func (r *repositoryImpl) listAll(ctx context.Context) ([]*Network, error) { rows, err := r.pool.Query(ctx, - `SELECT id, name, admin_human_id, created_at FROM networks`, + `SELECT `+networkColumns+` FROM networks`, ) if err != nil { return nil, err @@ -215,7 +244,7 @@ func (r *repositoryImpl) listAll(ctx context.Context) ([]*Network, error) { var networks []*Network for rows.Next() { var n Network - if err := rows.Scan(&n.ID, &n.Name, &n.AdminHumanId, &n.CreatedAt); err != nil { + if err := scanNetwork(rows, &n); err != nil { return nil, err } networks = append(networks, &n) @@ -293,8 +322,8 @@ func (r *repositoryImpl) getInvitationsByNetwork(ctx context.Context, networkID return invitations, rows.Err() } -func (r *repositoryImpl) deleteInvitation(ctx context.Context, networkID, email string) error { - _, err := r.pool.Exec(ctx, +func (r *repositoryImpl) deleteInvitation(ctx context.Context, db dbtx, networkID, email string) error { + _, err := db.Exec(ctx, `DELETE FROM network_invitations WHERE network_id = $1 AND email = $2`, networkID, email, ) diff --git a/go/internal/network/service.go b/go/internal/network/service.go index 738ed21..c06392a 100644 --- a/go/internal/network/service.go +++ b/go/internal/network/service.go @@ -9,8 +9,11 @@ import ( "strings" pbaero "github.com/flowy-live/llink/genproto/aero" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/utils" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "slices" ) var ErrNotFound = errors.New("network not found") @@ -25,8 +28,11 @@ type Service interface { GetByID(ctx context.Context, id string) (*Network, error) // SetName returns ErrNotFound or ErrInvalidName. SetName(ctx context.Context, id, name string) error + // AddMembers inserts members and syncs the new seat count to billing + // atomically; a Stripe failure rolls the insert back. AddMembers(ctx context.Context, networkID string, humanIds []string) error RemoveMember(ctx context.Context, networkID, humanId string) error + CountSeats(ctx context.Context, networkID string) (int, error) ListForHuman(ctx context.Context, humanId string) ([]*Network, error) IsMember(ctx context.Context, networkID, humanId string) (bool, error) // ListAll returns all networks with their members @@ -41,12 +47,19 @@ type Service interface { } type serviceImpl struct { - repo repository - aeroSvc pbaero.PrimaryClient + pool *pgxpool.Pool + repo repository + aeroSvc pbaero.PrimaryClient + billingSvc billing.Service } -func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient) Service { - return &serviceImpl{repo: newRepository(pool), aeroSvc: aeroSvc} +func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service) Service { + return &serviceImpl{ + pool: pool, + repo: newRepository(pool), + aeroSvc: aeroSvc, + billingSvc: billingSvc, + } } func (s *serviceImpl) Create(ctx context.Context, name, adminHumanId string) (*Network, error) { @@ -60,8 +73,7 @@ func (s *serviceImpl) Create(ctx context.Context, name, adminHumanId string) (*N return nil, err } - err = s.AddMembers(ctx, network.ID, []string{adminHumanId}) - if err != nil { + if err := s.AddMembers(ctx, network.ID, []string{adminHumanId}); err != nil { return nil, err } @@ -90,22 +102,58 @@ func (s *serviceImpl) SetName(ctx context.Context, id, name string) error { } func (s *serviceImpl) AddMembers(ctx context.Context, networkID string, humanIds []string) error { - for _, humanId := range humanIds { - if humanId == "" { - return fmt.Errorf("invalid humanId") - } - if err := s.repo.addMember(ctx, networkID, humanId); err != nil { - return err - } + if slices.Contains(humanIds, "") { + return fmt.Errorf("invalid humanId") } - return nil + return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + for _, humanId := range humanIds { + if err := s.repo.addMember(ctx, tx, networkID, humanId); err != nil { + return err + } + } + return nil + }) } func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId string) error { if humanId == "" { return fmt.Errorf("invalid humanId") } - return s.repo.removeMember(ctx, networkID, humanId) + return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + return s.repo.removeMember(ctx, tx, networkID, humanId) + }) +} + +// mutateMembers runs fn in a tx, recounts seats, calls billing.SyncSeats, +// and commits. Any error rolls the membership change 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 { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback(ctx) + + if err := fn(tx); err != nil { + return err + } + + seats, err := s.repo.countSeats(ctx, tx, networkID) + if err != nil { + return fmt.Errorf("count seats: %w", err) + } + + if err := s.billingSvc.SyncSeats(ctx, networkID, seats); err != nil { + return fmt.Errorf("sync billing seats: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit tx: %w", err) + } + return nil +} + +func (s *serviceImpl) CountSeats(ctx context.Context, networkID string) (int, error) { + return s.repo.countSeats(ctx, s.pool, networkID) } func (s *serviceImpl) ListForHuman(ctx context.Context, humanId string) ([]*Network, error) { @@ -126,8 +174,6 @@ func (s *serviceImpl) ListAll(ctx context.Context) ([]*Network, error) { return s.repo.listAll(ctx) } -// Invitation methods - func (s *serviceImpl) InviteByEmail(ctx context.Context, networkID string, emails []string) error { network, err := s.repo.getByID(ctx, networkID) if err != nil { @@ -180,10 +226,13 @@ func (s *serviceImpl) AcceptInvitation(ctx context.Context, networkID, email, hu return fmt.Errorf("invalid humanId") } - if err := s.repo.deleteInvitation(ctx, networkID, normalized); err != nil { - return err - } - return s.repo.addMember(ctx, networkID, humanId) + return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + err := s.repo.deleteInvitation(ctx, tx, networkID, normalized) + if err != nil { + return err + } + return s.repo.addMember(ctx, tx, networkID, humanId) + }) } func (s *serviceImpl) RevokeInvitation(ctx context.Context, networkID, email string) error { @@ -191,7 +240,7 @@ func (s *serviceImpl) RevokeInvitation(ctx context.Context, networkID, email str if err != nil { return fmt.Errorf("invalid email: %w", err) } - return s.repo.deleteInvitation(ctx, networkID, normalized) + return s.repo.deleteInvitation(ctx, s.pool, networkID, normalized) } func buildInvitationHTML(networkName string) string { diff --git a/go/internal/network/service_test.go b/go/internal/network/service_test.go index b2f2032..c14e4fa 100644 --- a/go/internal/network/service_test.go +++ b/go/internal/network/service_test.go @@ -6,6 +6,7 @@ import ( "testing" pbaero "github.com/flowy-live/llink/genproto/aero" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/network" "github.com/flowy-live/llink/internal/network/mocks" "github.com/flowy-live/llink/internal/testhelper" @@ -31,7 +32,7 @@ func newTestService(t *testing.T) network.Service { ShootEmail(gomock.Any(), gomock.Any()). Return(&pbaero.ShootEmailResponse{}, nil). AnyTimes() - return network.NewService(dbPool, mockAero) + return network.NewService(dbPool, mockAero, billing.Noop()) } func TestNetworkService(t *testing.T) { diff --git a/go/internal/particle/service_test.go b/go/internal/particle/service_test.go index b85ac72..61a6946 100644 --- a/go/internal/particle/service_test.go +++ b/go/internal/particle/service_test.go @@ -6,6 +6,7 @@ import ( "os" "testing" + "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/network" "github.com/flowy-live/llink/internal/particle" "github.com/flowy-live/llink/internal/testhelper" @@ -33,7 +34,7 @@ func getStreamStatus(data json.RawMessage) string { func TestParticleService_CreateAndGet(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network first @@ -72,7 +73,7 @@ func TestParticleService_CreateAndGet(t *testing.T) { func TestParticleService_NestedParticles(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -118,7 +119,7 @@ func TestParticleService_NestedParticles(t *testing.T) { func TestParticleService_CustomVisibility(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network with a member @@ -160,7 +161,7 @@ func TestParticleService_CustomVisibility(t *testing.T) { func TestParticleService_UpdateAndDelete(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -198,7 +199,7 @@ func TestParticleService_UpdateAndDelete(t *testing.T) { func TestParticleService_ListRootParticles(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -224,7 +225,7 @@ func TestParticleService_ListRootParticles(t *testing.T) { func TestParticleService_OpenCloseStream(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -272,7 +273,7 @@ func TestParticleService_OpenCloseStream(t *testing.T) { func TestParticleService_NotAStream(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -301,7 +302,7 @@ func TestParticleService_NotAStream(t *testing.T) { func TestParticleService_AccessInheritance(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil) + networkSvc := network.NewService(dbPool, nil, billing.Noop()) svc := particle.NewService(dbPool, networkSvc) // Create a network with members diff --git a/go/k8s/dev/orion.yaml b/go/k8s/dev/orion.yaml index fe5a099..0596ed6 100644 --- a/go/k8s/dev/orion.yaml +++ b/go/k8s/dev/orion.yaml @@ -61,6 +61,24 @@ spec: secretKeyRef: name: shared-secrets key: LIVEKIT_URL + - name: "STRIPE_SECRET_KEY" + valueFrom: + secretKeyRef: + name: shared-secrets + key: STRIPE_SECRET_KEY + - name: "STRIPE_WEBHOOK_SECRET" + valueFrom: + secretKeyRef: + name: shared-secrets + key: STRIPE_WEBHOOK_SECRET + - name: "STRIPE_PRICE_PRO_MONTHLY" + value: "price_1TM9abF9Z3lE6HEQJQmVsWES" + - name: "STRIPE_PRICE_PRO_ANNUAL" + value: "price_1TM9abF9Z3lE6HEQsToXflbq" + - name: "BILLING_SUCCESS_URL" + value: "llink://billing/success" + - name: "BILLING_CANCEL_URL" + value: "llink://billing/cancel" --- diff --git a/go/k8s/prod/orion.yaml b/go/k8s/prod/orion.yaml index 50b4e7f..aaee2e2 100644 --- a/go/k8s/prod/orion.yaml +++ b/go/k8s/prod/orion.yaml @@ -58,6 +58,24 @@ spec: secretKeyRef: name: shared-secrets key: LIVEKIT_URL + - name: "STRIPE_SECRET_KEY" + valueFrom: + secretKeyRef: + name: shared-secrets + key: STRIPE_SECRET_KEY + - name: "STRIPE_WEBHOOK_SECRET" + valueFrom: + secretKeyRef: + name: shared-secrets + key: STRIPE_WEBHOOK_SECRET + - name: "STRIPE_PRICE_PRO_MONTHLY" + value: "price_1TMBElJu6RWBXAm2pPthUoh6" + - name: "STRIPE_PRICE_PRO_ANNUAL" + value: "price_1TMBElJu6RWBXAm2TOpzdXk5" + - name: "BILLING_SUCCESS_URL" + value: "llink://billing/success" + - name: "BILLING_CANCEL_URL" + value: "llink://billing/cancel" --- diff --git a/go/migrations/000013_network_billing.down.sql b/go/migrations/000013_network_billing.down.sql new file mode 100644 index 0000000..7431273 --- /dev/null +++ b/go/migrations/000013_network_billing.down.sql @@ -0,0 +1,6 @@ +BEGIN; + +DROP TABLE IF EXISTS network_subscriptions; +DROP TABLE IF EXISTS network_stripe_customers; + +COMMIT; diff --git a/go/migrations/000013_network_billing.up.sql b/go/migrations/000013_network_billing.up.sql new file mode 100644 index 0000000..176a95a --- /dev/null +++ b/go/migrations/000013_network_billing.up.sql @@ -0,0 +1,28 @@ +BEGIN; + +CREATE TABLE network_stripe_customers ( + network_id TEXT PRIMARY KEY, + stripe_customer_id TEXT NOT NULL UNIQUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE network_subscriptions ( + id TEXT PRIMARY KEY, + network_id TEXT NOT NULL REFERENCES network_stripe_customers(network_id) ON DELETE CASCADE, + stripe_customer_id TEXT NOT NULL, + status TEXT NOT NULL, + price_id TEXT NOT NULL, + cadence TEXT NOT NULL, + quantity INT NOT NULL, + cancel_at_period_end BOOLEAN NOT NULL DEFAULT FALSE, + current_period_start TIMESTAMPTZ NOT NULL, + current_period_end TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT network_subscriptions_cadence_check CHECK (cadence IN ('monthly', 'annual')) +); + +CREATE UNIQUE INDEX idx_network_subscriptions_network_id + ON network_subscriptions (network_id); + +COMMIT; diff --git a/go/migrations/000014_network_message_usage.down.sql b/go/migrations/000014_network_message_usage.down.sql new file mode 100644 index 0000000..2364ec5 --- /dev/null +++ b/go/migrations/000014_network_message_usage.down.sql @@ -0,0 +1,5 @@ +BEGIN; + +DROP TABLE IF EXISTS network_message_usage; + +COMMIT; diff --git a/go/migrations/000014_network_message_usage.up.sql b/go/migrations/000014_network_message_usage.up.sql new file mode 100644 index 0000000..a6d082e --- /dev/null +++ b/go/migrations/000014_network_message_usage.up.sql @@ -0,0 +1,11 @@ +BEGIN; + +CREATE TABLE network_message_usage ( + network_id TEXT NOT NULL REFERENCES networks(id) ON DELETE CASCADE, + usage_date DATE NOT NULL, + message_count INTEGER NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (network_id, usage_date) +); + +COMMIT; diff --git a/js/src/api/client.ts b/js/src/api/client.ts index 24139ba..d1671e0 100644 --- a/js/src/api/client.ts +++ b/js/src/api/client.ts @@ -2,18 +2,23 @@ import { appConfig } from "@/config/env"; import { useSessionStore } from "@/stores/session-store"; import type { z } from "zod"; import { + BillingStatusSchema, + CheckoutSessionResponseSchema, DepotObjectSchema, GetLivekitTokenResponseSchema, HumanSchema, ListInvitationsResponseSchema, ListNetworksResponseSchema, NetworkSchema, + NetworkUsageSchema, + PortalSessionResponseSchema, PrepareUploadResponseSchema, SignInResponseSchema, } from "./types"; import type { AcceptInvitationRequest, AddMembersRequest, + BillingCadence, CreateNetworkRequest, PrepareUploadRequest, RequestCodeRequest, @@ -216,6 +221,41 @@ class ApiClient { async getLivekitToken(networkId: string, streamId: string) { return this.request(GetLivekitTokenResponseSchema, "POST", "/livekit/token", { network_id: networkId, stream_id: streamId }); } + + // --- Billing (network admin only) --- + + async getNetworkBilling(networkId: string) { + return this.request( + BillingStatusSchema, + "GET", + `/networks/${networkId}/billing`, + ); + } + + async createCheckoutSession(networkId: string, cadence: BillingCadence) { + return this.request( + CheckoutSessionResponseSchema, + "POST", + `/networks/${networkId}/billing/checkout-session`, + { cadence }, + ); + } + + async createPortalSession(networkId: string) { + return this.request( + PortalSessionResponseSchema, + "POST", + `/networks/${networkId}/billing/portal-session`, + ); + } + + async getNetworkUsage(networkId: string) { + return this.request( + NetworkUsageSchema, + "GET", + `/networks/${networkId}/usage`, + ); + } } export const apiClient = new ApiClient({ diff --git a/js/src/api/types.ts b/js/src/api/types.ts index 5e1ae03..d1b1a60 100644 --- a/js/src/api/types.ts +++ b/js/src/api/types.ts @@ -262,3 +262,53 @@ export const SignInResponseSchema = z.object({ token: z.string(), }); export type SignInResponse = z.infer; + +// --- Billing types --- + +export const BillingCadenceSchema = z.enum(["monthly", "annual"]); +export type BillingCadence = z.infer; + +export const NetworkPlanSchema = z.enum(["free", "pro"]); +export type NetworkPlan = z.infer; + +// Mirrors Stripe subscription.status plus "active" as the default free-tier value. +export const BillingPlanStatusSchema = z.enum([ + "active", + "trialing", + "past_due", + "canceled", + "incomplete", + "incomplete_expired", + "unpaid", +]); +export type BillingPlanStatus = z.infer; + +export const BillingStatusSchema = z.object({ + plan: NetworkPlanSchema, + plan_status: BillingPlanStatusSchema, + cadence: BillingCadenceSchema.nullable(), + seats: z.number().int(), + current_period_end: z.coerce.date().nullable(), + cancel_at_period_end: z.boolean(), + price_monthly_cents: z.number().int(), + price_annual_cents: z.number().int(), +}); +export type BillingStatus = z.infer; + +export const CheckoutSessionResponseSchema = z.object({ + url: z.string().url(), +}); +export type CheckoutSessionResponse = z.infer; + +export const PortalSessionResponseSchema = z.object({ + url: z.string().url(), +}); +export type PortalSessionResponse = z.infer; + +export const NetworkUsageSchema = z.object({ + plan: NetworkPlanSchema, + used: z.number().int().nonnegative(), + limit: z.number().int().nonnegative().nullable(), + reset_at: z.coerce.date(), +}); +export type NetworkUsage = z.infer; diff --git a/js/src/components/ui/radio-group.tsx b/js/src/components/ui/radio-group.tsx new file mode 100644 index 0000000..6bb6b5d --- /dev/null +++ b/js/src/components/ui/radio-group.tsx @@ -0,0 +1,42 @@ +import * as React from "react" +import { RadioGroup as RadioGroupPrimitive } from "radix-ui" + +import { cn } from "@/lib/utils" + +function RadioGroup({ + className, + ...props +}: React.ComponentProps) { + return ( + + ) +} + +function RadioGroupItem({ + className, + ...props +}: React.ComponentProps) { + return ( + + + + + + ) +} + +export { RadioGroup, RadioGroupItem } diff --git a/js/src/features/compose/compose-overlay.tsx b/js/src/features/compose/compose-overlay.tsx index 8713b5d..d361902 100644 --- a/js/src/features/compose/compose-overlay.tsx +++ b/js/src/features/compose/compose-overlay.tsx @@ -1,7 +1,8 @@ import { useCallback, useEffect, useEffectEvent, useRef, useState } from "react"; import { toast } from "sonner"; import { useAuthStore } from "@/stores/auth-store"; -import { useCreateParticle, useCreateStreamParticle } from "@/hooks/use-create-particle"; +import { QuotaExceededError, useCreateParticle, useCreateStreamParticle } from "@/hooks/use-create-particle"; +import { isUsageExhausted, useInvalidateNetworkUsage, useNetworkUsage } from "@/hooks/use-network-usage"; import { useRecorder } from "@/features/compose/use-recorder"; import { useScreenRecorder } from "@/features/compose/use-screen-recorder"; import { particlePath, parseParticlePath } from "@/lib/particle-path"; @@ -70,12 +71,17 @@ export function ComposeOverlay({ const userId = useAuthStore((s) => s.user?.id); const createParticle = useCreateParticle(); const createStream = useCreateStreamParticle(); + const { data: usage } = useNetworkUsage(networkId); + const invalidateUsage = useInvalidateNetworkUsage(); + const quotaExhausted = isUsageExhausted(usage); // Refs for synchronous reads in keyboard handlers const stepRef = useRef(step); const recordStartRef = useRef(0); const disabledRef = useRef(disabled); disabledRef.current = disabled; + const quotaExhaustedRef = useRef(quotaExhausted); + quotaExhaustedRef.current = quotaExhausted; const recordingSourceRef = useRef(recordingSource); recordingSourceRef.current = recordingSource; @@ -88,7 +94,12 @@ export function ComposeOverlay({ useEffect(() => { onActiveChange?.(step !== "idle"); onStepChange?.(step); - }, [step, onActiveChange, onStepChange]); + // Refresh quota when the overlay activates — user is about to send, so + // we want the most accurate count before the client-side gate kicks in. + if (step !== "idle") { + void invalidateUsage(networkId); + } + }, [step, onActiveChange, onStepChange, invalidateUsage, networkId]); const revokeAttachmentThumbnails = useCallback((items: PendingAttachment[]) => { for (const a of items) { @@ -334,12 +345,25 @@ export function ComposeOverlay({ ], ); + const handleQuotaError = useCallback((err: unknown): boolean => { + if (err instanceof QuotaExceededError) { + toast.error("Daily message limit reached. Upgrade to Pro to keep sending."); + cancel(); + return true; + } + return false; + }, [cancel]); + // Reply mode: create particle directly under targetPath const onSubmitReply = useEffectEvent(async () => { if (!targetPath || !userId || stepRef.current === "submitting") return; setStepSync("submitting"); - await createChildParticle(targetPath); - cancel(); + try { + await createChildParticle(targetPath); + cancel(); + } catch (err) { + if (!handleQuotaError(err)) throw err; + } }); // New stream mode: create stream + first child @@ -348,21 +372,25 @@ export function ComposeOverlay({ if (!userId || stepRef.current === "submitting") return; setStepSync("submitting"); - const streamId = await createStream.mutateAsync({ - networkId, - properties: { - name: streamName, - }, - createdByHumanId: userId, - visibleTo, - }); + try { + const streamId = await createStream.mutateAsync({ + networkId, + properties: { + name: streamName, + }, + createdByHumanId: userId, + visibleTo, + }); - const streamChildrenPath = particlePath(networkId, [streamId]); - await createChildParticle(streamChildrenPath); + const streamChildrenPath = particlePath(networkId, [streamId]); + await createChildParticle(streamChildrenPath); - cancel(); + cancel(); + } catch (err) { + if (!handleQuotaError(err)) throw err; + } }, - [networkId, userId, createParticle, createChildParticle, cancel], + [networkId, userId, createStream, createChildParticle, cancel, handleQuotaError], ); // --- Keyboard handling --- @@ -397,6 +425,13 @@ export function ComposeOverlay({ } break; } + if (quotaExhaustedRef.current) { + if ((e.key === "`" && !e.repeat) || e.key === "t" || e.key === "T" || e.key === "s" || e.key === "S") { + e.preventDefault(); + toast.info("Daily message limit reached. Upgrade to Pro to keep sending."); + } + break; + } if (e.key === "`" && !e.repeat) { e.preventDefault(); recordStartRef.current = Date.now(); diff --git a/js/src/features/compose/compose-quota-indicator.tsx b/js/src/features/compose/compose-quota-indicator.tsx new file mode 100644 index 0000000..567ec93 --- /dev/null +++ b/js/src/features/compose/compose-quota-indicator.tsx @@ -0,0 +1,94 @@ +import { useNavigate } from "react-router-dom"; +import { Progress } from "@/components/ui/progress"; +import { Button } from "@/components/ui/button"; +import { useNetworkUsage } from "@/hooks/use-network-usage"; +import { useIsNetworkAdmin, useNetwork } from "@/hooks/use-networks"; + +interface ComposeQuotaIndicatorProps { + networkId: string; +} + +const SHOW_PROGRESS_AT_FRACTION = 0.7; + +/** + * Surfaces freemium quota state near compose: + * - Nothing below 70% used (avoid nagging). + * - A subtle progress pill between 70% and the limit. + * - A locked banner with an upgrade CTA once the limit is hit. + * + * Pro networks and any network still loading usage render nothing. + */ +export function ComposeQuotaIndicator({ networkId }: ComposeQuotaIndicatorProps) { + const navigate = useNavigate(); + const { data: usage } = useNetworkUsage(networkId); + const isAdmin = useIsNetworkAdmin(networkId); + const network = useNetwork(networkId); + + if (!usage || usage.limit == null) return null; + + const fraction = usage.used / usage.limit; + const exhausted = usage.used >= usage.limit; + + if (exhausted) { + return ( +
+
+ {isAdmin + ? `You've reached today's ${usage.limit}-message limit` + : `This network reached today's ${usage.limit}-message limit`} +
+
+ Resets {formatResetRelative(usage.reset_at)} ({formatResetAbsolute(usage.reset_at)}) +
+ {isAdmin ? ( + + ) : ( +
+ Ask{" "} + + {network?.admin_human.email_prefix ?? "your admin"} + {" "} + to upgrade to Pro +
+ )} +
+ ); + } + + if (fraction < SHOW_PROGRESS_AT_FRACTION) return null; + + return ( +
+ + {usage.used}/{usage.limit} today + + +
+ ); +} + +function formatResetRelative(resetAt: Date): string { + const now = new Date(); + const diffMs = resetAt.getTime() - now.getTime(); + const hours = Math.max(0, Math.round(diffMs / (60 * 60 * 1000))); + if (hours < 1) return "soon"; + if (hours === 1) return "in 1 hour"; + return `in ${hours} hours`; +} + +function formatResetAbsolute(resetAt: Date): string { + // Shows the user their local wall-clock time for the UTC-midnight reset, + // so a user in UTC-8 sees "4:00 PM" instead of a relative hint alone. + return resetAt.toLocaleTimeString(undefined, { + hour: "numeric", + minute: "2-digit", + }); +} diff --git a/js/src/features/network-billing.tsx b/js/src/features/network-billing.tsx new file mode 100644 index 0000000..42df501 --- /dev/null +++ b/js/src/features/network-billing.tsx @@ -0,0 +1,339 @@ +import { useState } from "react"; +import { ExternalLink } from "lucide-react"; +import { toast } from "sonner"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Label } from "@/components/ui/label"; +import { RadioGroup, RadioGroupItem } from "@/components/ui/radio-group"; +import { Separator } from "@/components/ui/separator"; +import { Muted } from "@/components/ui/typography"; +import { cn } from "@/lib/utils"; +import { + useCreateCheckoutSession, + useCreatePortalSession, + useNetworkBilling, +} from "@/hooks/use-billing"; +import { useNetworkUsage } from "@/hooks/use-network-usage"; +import { useIsNetworkAdmin } from "@/hooks/use-networks"; +import type { BillingCadence, BillingStatus } from "@/api/types"; + +function formatCents(cents: number): string { + if (cents % 100 === 0) return `$${cents / 100}`; + return `$${(cents / 100).toFixed(2)}`; +} + +function formatDate(date: Date): string { + return date.toLocaleDateString(undefined, { + month: "long", + day: "numeric", + year: "numeric", + }); +} + +function PlanStatusBadge({ status }: { status: BillingStatus["plan_status"] }) { + if (status === "past_due") + return Past due; + if (status === "canceled") return Canceled; + if (status === "trialing") return Trialing; + return null; +} + +function InfoRow({ + label, + value, +}: { + label: React.ReactNode; + value: React.ReactNode; +}) { + return ( +
+ {label} +
+
{value}
+
+ ); +} + +function CadenceOption({ + value, + label, + perSeatCents, + billedNote, + saveBadge, + selected, +}: { + value: BillingCadence; + label: string; + perSeatCents: number; + billedNote: string; + saveBadge?: string; + selected: boolean; +}) { + return ( + + ); +} + +function formatResetLocal(resetAt: Date): string { + const now = new Date(); + const diffMs = resetAt.getTime() - now.getTime(); + const hours = Math.max(0, Math.round(diffMs / (60 * 60 * 1000))); + const absolute = resetAt.toLocaleTimeString(undefined, { + hour: "numeric", + minute: "2-digit", + }); + if (hours < 1) return `soon (${absolute})`; + if (hours === 1) return `in 1 hour (${absolute})`; + return `in ${hours} hours (${absolute})`; +} + +/** + * Read-only plan + quota summary, sourced from `/usage` (member-accessible). + * The `/billing` endpoint is admin-gated, so we can't use it for the + * everyone-visible summary. + */ +function PlanSummary({ networkId }: { networkId: string }) { + const { data: usage } = useNetworkUsage(networkId); + + if (!usage) return null; + + const isPro = usage.plan === "pro"; + + return ( + <> + + {isPro ? "Llink Pro" : "Llink Free"} + + {isPro ? "Pro" : "Free"} + +
+ } + /> + {!isPro && ( + <> + + + + {usage.used} / {usage.limit} + + + Resets {formatResetLocal(usage.reset_at)} + + + ) : ( + — + ) + } + /> + + )} + + ); +} + +function FreeBilling({ + networkId, + billing, +}: { + networkId: string; + billing: BillingStatus; +}) { + const createCheckout = useCreateCheckoutSession(networkId); + const [cadence, setCadence] = useState("annual"); + + const handleUpgrade = () => { + createCheckout.mutate(cadence, { + onSuccess: ({ url }) => window.electronLink.openExternal(url), + onError: (err) => toast.error(err.message || "Failed to start checkout"), + }); + }; + + const annualPerSeatMonthlyCents = Math.round(billing.price_annual_cents / 12); + const savingsPct = Math.round( + (1 - annualPerSeatMonthlyCents / billing.price_monthly_cents) * 100, + ); + + return ( + <> + setCadence(v as BillingCadence)} + className="gap-0" + > + 0 ? `Save ${savingsPct}%` : undefined} + selected={cadence === "annual"} + /> + + + +
+ +
+ + ); +} + +function ProBilling({ + networkId, + billing, +}: { + networkId: string; + billing: BillingStatus; +}) { + const createPortal = useCreatePortalSession(networkId); + + const handleManage = () => { + createPortal.mutate(undefined, { + onSuccess: ({ url }) => window.electronLink.openExternal(url), + onError: (err) => + toast.error(err.message || "Failed to open billing portal"), + }); + }; + + const cadenceLabel = billing.cadence === "annual" ? "Annual" : "Monthly"; + const perSeatCents = + billing.cadence === "annual" + ? Math.round(billing.price_annual_cents / 12) + : billing.price_monthly_cents; + const renewal = billing.current_period_end + ? formatDate(billing.current_period_end) + : null; + + return ( + <> + {billing.cancel_at_period_end && renewal && ( +
+ Your subscription is set to downgrade to Free on {renewal}. +
+ )} + {billing.plan_status === "past_due" && ( +
+ Your last payment failed. Update your payment method to keep Pro + active. +
+ )} + + + {`${cadenceLabel} · ${formatCents(perSeatCents)} / seat / mo`} + + + } + /> + + + {renewal && ( + <> + + + + )} +
+ +
+ + ); +} + +/** + * Unified billing section. Shows the plan + usage summary to every member, + * and the admin-only management controls (upgrade / portal) below. + * + * `/billing` is admin-gated, so the management controls are the only part + * that depends on it — members rely on `/usage` for the summary. + */ +export function BillingSection({ networkId }: { networkId: string }) { + const isAdmin = useIsNetworkAdmin(networkId); + return ( + <> + + {isAdmin && ( + <> + + + + )} + + ); +} + +function AdminBillingControls({ networkId }: { networkId: string }) { + const { data: billing, isLoading, error } = useNetworkBilling(networkId); + + if (isLoading || !billing) { + return ( +
+ Loading billing... +
+ ); + } + if (error) { + return ( +
+ Failed to load billing. +
+ ); + } + if (billing.plan === "pro") { + return ; + } + return ; +} diff --git a/js/src/features/network-root.tsx b/js/src/features/network-root.tsx index 64c7087..a1d4af8 100644 --- a/js/src/features/network-root.tsx +++ b/js/src/features/network-root.tsx @@ -5,6 +5,7 @@ import { particlePath } from "@/lib/particle-path"; import { ParticleListView } from "@/features/particles/particle-list-view"; import { VideoAudioToggle } from "@/components/video-audio-toggle"; import { ComposeOverlay } from "./compose/compose-overlay"; +import { ComposeQuotaIndicator } from "./compose/compose-quota-indicator"; import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { useStreamParticles } from "@/hooks/use-stream-particles"; import { useStreamKeyboardNav } from "@/hooks/use-stream-keyboard-nav"; @@ -50,7 +51,7 @@ export default function NetworkRoot() { return (
{/* Top bar — stays in place */} -
+
setStatusTab(v === "closed" ? "closed" : "open")} @@ -75,6 +76,11 @@ export default function NetworkRoot() {
+ {!composeActive && ( +
+ +
+ )}
diff --git a/js/src/features/network-settings.tsx b/js/src/features/network-settings.tsx index b2ca79d..f44ddbf 100644 --- a/js/src/features/network-settings.tsx +++ b/js/src/features/network-settings.tsx @@ -1,6 +1,6 @@ -import { useCallback, useEffect, useRef, useState } from "react"; -import { useNavigate, useParams } from "react-router-dom"; -import { ArrowLeft, Mail, Shield, X } from "lucide-react"; +import { useEffect, useRef, useState } from "react"; +import { useNavigate, useParams, useSearchParams } from "react-router-dom"; +import { ArrowLeft, CreditCard, Mail, Shield, Users, X } from "lucide-react"; import { toast } from "sonner"; import { Avatar, AvatarFallback } from "@/components/ui/avatar"; import { Badge } from "@/components/ui/badge"; @@ -17,15 +17,10 @@ import { useRevokeInvitation, } from "@/hooks/use-invitations"; import { useAuthStore } from "@/stores/auth-store"; +import { BillingSection } from "@/features/network-billing"; import type { Human } from "@/api/types"; -function MemberRow({ - human, - isAdmin, -}: { - human: Human; - isAdmin: boolean; -}) { +function MemberRow({ human, isAdmin }: { human: Human; isAdmin: boolean }) { const initials = human.email_prefix.slice(0, 2).toUpperCase(); return ( @@ -73,7 +68,7 @@ function InviteForm({ networkId }: { networkId: string }) {
setEmail(e.target.value)} className="flex-1" @@ -111,7 +106,7 @@ function PendingInvitationRow({ return (
- +
@@ -131,33 +126,63 @@ function PendingInvitationRow({ ); } -function SettingsGroup({ +function SectionHeader({ + icon, title, - children, + description, + trailing, }: { + icon: React.ReactNode; title: string; - children: React.ReactNode; + description?: string; + trailing?: React.ReactNode; }) { return ( -
-

- {title} -

-
{children}
+
+ + {icon} + +
+
+

{title}

+ {trailing} +
+ {description && {description}} +
); } +function Section({ children }: { children: React.ReactNode }) { + return ( +
+ {children} +
+ ); +} + export default function NetworkSettingsPage() { const navigate = useNavigate(); const { networkId } = useParams<{ networkId: string }>(); + const [searchParams] = useSearchParams(); const { data: networks } = useNetworks(); const network = networks?.find((n) => n.id === networkId); const { data: invitations } = useNetworkInvitations(networkId!); const currentUser = useAuthStore((s) => s.user); const isAdmin = currentUser?.id === network?.admin_human.id; + const billingRef = useRef(null); + + useEffect(() => { + if (searchParams.get("section") === "billing") { + billingRef.current?.scrollIntoView({ behavior: "smooth", block: "start" }); + } + }, [searchParams]); + const networkName = network?.name ?? "Network"; + const memberCount = network?.humans.length ?? 0; + const pendingCount = invitations?.length ?? 0; + const networkInitials = networkName.slice(0, 2).toUpperCase(); return (
@@ -171,12 +196,38 @@ export default function NetworkSettingsPage() { > - {networkName} + Settings
- - + +
+ + + {networkInitials} + + +
+

{networkName}

+ + {memberCount} {memberCount === 1 ? "member" : "members"} + {isAdmin ? " · You're an admin" : ""} + +
+
+ +
+ } + title="Members" + description="People with access to this network." + trailing={ + + {memberCount} + + } + /> + {network?.humans.map((human, index) => (
))} - - - +
{isAdmin && network && ( - <> - - - - - - - - {invitations && invitations.length > 0 ? ( - invitations.map((inv, index) => ( +
+ } + title="Invitations" + description="Invite teammates by email. They'll get a link to join." + trailing={ + pendingCount > 0 ? ( + + {pendingCount} pending + + ) : undefined + } + /> + + + {pendingCount > 0 && ( + <> + +
+ + Pending + +
+ {invitations!.map((inv, index) => (
- {index < invitations.length - 1 && ( + {index < invitations!.length - 1 && ( )}
- )) - ) : ( -

- No pending invitations -

- )} - - + ))} + + )} +
)} + +
+
+ } + title="Billing" + description={ + isAdmin + ? "Manage your plan, seats, and payment." + : "Your network's current plan and usage." + } + /> + + +
+
+ +
); diff --git a/js/src/features/particles/particle-list-view.tsx b/js/src/features/particles/particle-list-view.tsx index 529fd20..ae9d1a4 100644 --- a/js/src/features/particles/particle-list-view.tsx +++ b/js/src/features/particles/particle-list-view.tsx @@ -28,8 +28,38 @@ import { isParticleDeleted, type Particle, type StreamProperties } from "@/api/t import type { StreamParticle } from "@/hooks/use-stream-particles"; import { useNetwork } from "@/hooks/use-networks"; import { useStreamAutoplay } from "@/hooks/use-stream-autoplay"; +import { useDownloadUrl } from "@/hooks/use-download-url"; import { StreamContextMenu } from "@/features/particles/stream-context-menu"; +function VideoThumbnail({ + objectId, + isUnseen, +}: { + objectId: string; + isUnseen: boolean; +}) { + const { data: url } = useDownloadUrl(objectId); + return ( +
+ {url && ( +
+ ); +} + function getParticleTypeIcon(particle: Particle): LucideIcon { if (isParticleDeleted(particle)) return Trash2; switch (particle.type) { @@ -156,6 +186,14 @@ const StreamRow = memo(function StreamRow({ const TypeIcon = latestChild ? getParticleTypeIcon(latestChild) : Radio; + const videoThumbObjectId = + latestChild && + !isParticleDeleted(latestChild) && + latestChild.type === "media" && + latestChild.properties.mime_type.startsWith("video/") + ? latestChild.properties.object_id + : null; + return (
)} - - - {initials} - - + {videoThumbObjectId ? ( + + ) : ( + + + {initials} + + + )}

- +
{!permissionGranted && (
diff --git a/js/src/hooks/use-billing.ts b/js/src/hooks/use-billing.ts new file mode 100644 index 0000000..0ef195a --- /dev/null +++ b/js/src/hooks/use-billing.ts @@ -0,0 +1,29 @@ +import { useQuery, useMutation } from "@tanstack/react-query"; +import { apiClient } from "@/api/client"; +import type { BillingCadence } from "@/api/types"; + +export function useNetworkBilling(networkId: string | undefined) { + return useQuery({ + queryKey: ["network-billing", networkId], + queryFn: () => apiClient.getNetworkBilling(networkId!), + enabled: !!networkId, + // Refetch on window focus so the UI catches up after the user returns + // from Stripe Checkout (webhook may land a second or two later). + // FIX: doesn't work with electron + refetchOnWindowFocus: true, + refetchInterval: 10000 + }); +} + +export function useCreateCheckoutSession(networkId: string) { + return useMutation({ + mutationFn: (cadence: BillingCadence) => + apiClient.createCheckoutSession(networkId, cadence), + }); +} + +export function useCreatePortalSession(networkId: string) { + return useMutation({ + mutationFn: () => apiClient.createPortalSession(networkId), + }); +} diff --git a/js/src/hooks/use-create-particle.ts b/js/src/hooks/use-create-particle.ts index 66e8def..b49767e 100644 --- a/js/src/hooks/use-create-particle.ts +++ b/js/src/hooks/use-create-particle.ts @@ -1,7 +1,26 @@ -import { useMutation } from "@tanstack/react-query"; +import { useMutation, useQueryClient } from "@tanstack/react-query"; import { createParticle, createStreamParticle } from "@/lib/firestore-particles"; -import type { ParticleType, ParticlePropertiesMap } from "@/api/types"; -import { particlePath, ParticlePath, toFirestoreChildrenPath, toFirestoreDocPath } from "@/lib/particle-path"; +import { CONTAINER_TYPES, type NetworkUsage, type ParticleType, type ParticlePropertiesMap } from "@/api/types"; +import { parseParticlePath, particlePath, ParticlePath, toFirestoreChildrenPath } from "@/lib/particle-path"; +import { + isUsageExhausted, + networkUsageQueryKey, + useBumpNetworkUsage, + useInvalidateNetworkUsage, +} from "./use-network-usage"; + +/** + * Thrown when a free-plan network attempts to create a non-container particle + * after hitting its daily message limit. Callers should surface an upgrade + * prompt; compose UI should also disable triggers proactively via + * `useNetworkUsage` rather than relying on this throw. + */ +export class QuotaExceededError extends Error { + constructor(public readonly networkId: string) { + super("Daily message limit reached"); + this.name = "QuotaExceededError"; + } +} interface CreateParticleParams { // Path to which the new particle will be added as a child @@ -12,16 +31,37 @@ interface CreateParticleParams { } export function useCreateParticle() { + const qc = useQueryClient(); + const bumpUsage = useBumpNetworkUsage(); + const invalidateUsage = useInvalidateNetworkUsage(); + return useMutation({ mutationFn: async (params: CreateParticleParams) => { + const { networkId } = parseParticlePath(params.path); + + // Containers aren't counted server-side, so we don't block them. + if (!CONTAINER_TYPES.has(params.type)) { + const cached = qc.getQueryData(networkUsageQueryKey(networkId)); + if (isUsageExhausted(cached)) { + throw new QuotaExceededError(networkId); + } + } + const collectionPath = toFirestoreChildrenPath(params.path); - return await createParticle( + const result = await createParticle( collectionPath, params.type, params.properties, params.createdByHumanId, ); - } + + if (!CONTAINER_TYPES.has(params.type)) { + bumpUsage(networkId); + void invalidateUsage(networkId); + } + + return result; + }, }); } diff --git a/js/src/hooks/use-network-usage.ts b/js/src/hooks/use-network-usage.ts new file mode 100644 index 0000000..e88ab0b --- /dev/null +++ b/js/src/hooks/use-network-usage.ts @@ -0,0 +1,57 @@ +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { useCallback } from "react"; +import { apiClient } from "@/api/client"; +import type { NetworkUsage } from "@/api/types"; + +export const networkUsageQueryKey = (networkId: string | undefined) => + ["network-usage", networkId] as const; + +export function useNetworkUsage(networkId: string | undefined) { + return useQuery({ + queryKey: networkUsageQueryKey(networkId), + queryFn: () => apiClient.getNetworkUsage(networkId!), + enabled: !!networkId, + // Refetch whenever a consumer mounts (billing settings, compose indicator) + // so users land on fresh quota state without listener wiring. + refetchOnMount: "always", + refetchOnWindowFocus: true, + }); +} + +/** + * Returns a callback that invalidates the usage query for a network. + * Callers: own-send success path, inbound-particle listener. + */ +export function useInvalidateNetworkUsage() { + const qc = useQueryClient(); + return useCallback( + (networkId: string) => + qc.invalidateQueries({ queryKey: networkUsageQueryKey(networkId) }), + [qc], + ); +} + +/** + * Optimistic bump of the cached `used` count. The worker-written truth is + * reconciled on the next invalidation/refetch. + */ +export function useBumpNetworkUsage() { + const qc = useQueryClient(); + return useCallback( + (networkId: string) => { + qc.setQueryData(networkUsageQueryKey(networkId), (prev) => + prev ? { ...prev, used: prev.used + 1 } : prev, + ); + }, + [qc], + ); +} + +/** + * True iff the network is on the free plan and has exhausted today's quota. + */ +export function isUsageExhausted(usage: NetworkUsage | undefined): boolean { + if (!usage) return false; + if (usage.limit == null) return false; + return usage.used >= usage.limit; +} diff --git a/js/src/hooks/use-networks.ts b/js/src/hooks/use-networks.ts index 7e2174d..1608023 100644 --- a/js/src/hooks/use-networks.ts +++ b/js/src/hooks/use-networks.ts @@ -1,5 +1,6 @@ import { useQuery } from "@tanstack/react-query"; import { apiClient } from "@/api/client"; +import { useAuthStore } from "@/stores/auth-store"; export function useNetworks() { return useQuery({ @@ -12,3 +13,10 @@ export function useNetwork(networkId: string) { const { data: networks } = useNetworks(); return networks?.find((n) => n.id === networkId) || null; } + +export function useIsNetworkAdmin(networkId: string): boolean { + const network = useNetwork(networkId); + const userId = useAuthStore((s) => s.user?.id); + if (!network || !userId) return false; + return network.admin_human.id === userId; +} diff --git a/js/src/hooks/use-stream-playback.ts b/js/src/hooks/use-stream-playback.ts index 7cb8758..5efb5bc 100644 --- a/js/src/hooks/use-stream-playback.ts +++ b/js/src/hooks/use-stream-playback.ts @@ -2,7 +2,7 @@ import { useCallback, useEffect, useEffectEvent, useMemo, useReducer, useRef, us import { useAuthStore } from "@/stores/auth-store"; import type { Particle } from "@/api/types"; import { useLiveParticleChildren } from "@/hooks/use-particle"; -import { parseParticlePath, toFirestoreDocPath, type ParticlePath } from "@/lib/particle-path"; +import { toFirestoreDocPath, type ParticlePath } from "@/lib/particle-path"; import { updateStreamPlaybackMarker } from "@/lib/firestore-particles"; // --- Playback reducer (ID-based) --- diff --git a/js/src/main.ts b/js/src/main.ts index 9c3ed10..2261477 100644 --- a/js/src/main.ts +++ b/js/src/main.ts @@ -25,11 +25,21 @@ if (process.platform === 'darwin' && !app.isPackaged) { app.dock?.setIcon(path.join(__dirname, '../../assets/icon.png')); } +// In dev, `LLINK_PROFILE=foo yarn start` spins up a second instance with an +// isolated userData dir so it can coexist with the default one (separate auth, +// cookies, leveldb locks). +const devProfile = !app.isPackaged ? process.env.LLINK_PROFILE : undefined; +if (devProfile) { + app.setPath('userData', `${app.getPath('userData')}-${devProfile}`); +} + // Single-instance lock: on Windows/Linux, clicking a llink:// URL launches a new // process. The lock makes the losing instance quit and fires `second-instance` on // the primary, so we focus the existing window instead of spawning a duplicate. // macOS uses `open-url` instead and doesn't need this, but the lock is harmless. -if (!app.requestSingleInstanceLock()) { +// Skip the lock when running a named dev profile — those instances are meant to +// run alongside the default one. +if (!devProfile && !app.requestSingleInstanceLock()) { app.quit(); }