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

implement core foundation

Browse Source
This commit is contained in:
talksik
2026-04-14 11:00:46 -07:00
parent aff18d82db
commit 279a3e52c2
25 changed files with 1455 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/internal/network/models.go
+5 -5
View File
@@ -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 {
go/internal/network/repository.go
+54 -25
View File
@@ -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,
)
go/internal/network/service.go
+71 -26
View File
@@ -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")
@@ -19,17 +22,16 @@ var ErrCapacityExceeded = errors.New("active stream capacity exceeded")
var ErrInvalidRetentionHours = errors.New("message retention hours must be between 24 and 336")
type Service interface {
// Create creates a network and adds adminHumanId as the first member. Returns ErrInvalidName if name is empty.
Create(ctx context.Context, name, adminHumanId string) (*Network, error)
// GetByID returns ErrNotFound if network doesn't exist.
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
ListAll(ctx context.Context) ([]*Network, error)
// Invitations (email-based, for users who haven't registered yet)
@@ -41,12 +43,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 +69,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 +98,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 +170,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 +222,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 +236,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 {