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

implement paywall (#161)

Browse Source 
* implement core foundation

* inject deps

* fix incorrect migration

* tail migration

* use transaction for migration

* fix: inject deps for tests

* cleanup billing management for admin

* upgrade stripe sdk to v85

* set price env variables

* cleanup billing management

* allow multiple dev windows

* fix: settings scroll

* feat: show nice video thumbnail in listview

* feat: implement freemium restrictions

* remove unnecessary comments

* refactor

* docs

* format

* tweak network settings better hierarchy
This commit was merged in pull request #161.
This commit is contained in:
Arjun Patel
2026-04-14 15:18:32 -07:00
committed by GitHub
parent aff18d82db
commit 67826b92c0
44 changed files with 2197 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/cmd/emailnotifierjob/main.go
+2 -1
View File
@@ -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 {
go/cmd/orion/main.go
+26 -9
View File
@@ -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))
go/cmd/particleprocessorworker/main.go
+57
View File
@@ -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).
go/cmd/pusherservice/main.go
+3 -2
View File
@@ -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()