package particle import ( "context" "encoding/json" "errors" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "go.jetify.com/typeid" ) var errNotFound = errors.New("not found") var errAccessDenied = errors.New("access denied") type particleIDPrefix struct{} func (particleIDPrefix) Prefix() string { return "particle" } type particleID struct { typeid.TypeID[particleIDPrefix] } func newParticleID() (particleID, error) { return typeid.New[particleID]() } type repository interface { create(ctx context.Context, p *Particle) (*Particle, error) getByID(ctx context.Context, id string) (*Particle, error) update(ctx context.Context, id string, data json.RawMessage, updatedAt time.Time) error delete(ctx context.Context, id string) error list(ctx context.Context, networkID string, parentID *string, requesterEmail string, filter ListFilter, limit int, cursor *Cursor) ([]*Particle, error) setVisibility(ctx context.Context, id string, mode VisibilityMode) error addMembers(ctx context.Context, particleID string, emails []string) error removeMembers(ctx context.Context, particleID string, emails []string) error getMembers(ctx context.Context, particleID string) ([]string, error) getMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error) // getAncestorChain returns the particle and all its ancestors (for access checks) getAncestorChain(ctx context.Context, particleID string) ([]*Particle, error) isMemberOf(ctx context.Context, particleID, email string) (bool, error) // Seen tracking markSeen(ctx context.Context, particleID, email string) error getSeenMap(ctx context.Context, particleIDs []string, email string) (map[string]bool, error) getUnseenCounts(ctx context.Context, streamIDs []string, email string) (map[string]int, error) // Ack tracking ack(ctx context.Context, particleID, email string) error getAcksMap(ctx context.Context, particleIDs []string) (map[string][]AckInfo, error) } type repositoryImpl struct { pool *pgxpool.Pool } func newRepository(pool *pgxpool.Pool) repository { return &repositoryImpl{pool: pool} } func (r *repositoryImpl) create(ctx context.Context, p *Particle) (*Particle, error) { id, err := newParticleID() if err != nil { return nil, err } var result Particle err = r.pool.QueryRow(ctx, `INSERT INTO particles (id, type, network_id, parent_id, created_by_email, visibility, data) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id, type, network_id, parent_id, created_by_email, visibility, data, updated_at, created_at`, id.String(), p.Type, p.NetworkID, p.ParentID, p.CreatedByEmail, p.Visibility, p.Data, ).Scan(&result.ID, &result.Type, &result.NetworkID, &result.ParentID, &result.CreatedByEmail, &result.Visibility, &result.Data, &result.UpdatedAt, &result.CreatedAt) if err != nil { return nil, err } return &result, nil } func (r *repositoryImpl) getByID(ctx context.Context, id string) (*Particle, error) { var p Particle err := r.pool.QueryRow(ctx, `SELECT id, type, network_id, parent_id, created_by_email, visibility, data, updated_at, created_at FROM particles WHERE id = $1`, id, ).Scan(&p.ID, &p.Type, &p.NetworkID, &p.ParentID, &p.CreatedByEmail, &p.Visibility, &p.Data, &p.UpdatedAt, &p.CreatedAt) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, errNotFound } return nil, err } return &p, nil } func (r *repositoryImpl) update(ctx context.Context, id string, data json.RawMessage, updatedAt time.Time) error { result, err := r.pool.Exec(ctx, `UPDATE particles SET data = $1, updated_at = $2 WHERE id = $3`, data, updatedAt, id, ) if err != nil { return err } if result.RowsAffected() == 0 { return errNotFound } return nil } func (r *repositoryImpl) delete(ctx context.Context, id string) error { result, err := r.pool.Exec(ctx, `DELETE FROM particles WHERE id = $1`, id) if err != nil { return err } if result.RowsAffected() == 0 { return errNotFound } return nil } func (r *repositoryImpl) list(ctx context.Context, networkID string, parentID *string, requesterEmail string, filter ListFilter, limit int, cursor *Cursor) ([]*Particle, error) { query := `SELECT id, type, network_id, parent_id, created_by_email, visibility, data, updated_at, created_at FROM particles p WHERE p.network_id = $1` args := []any{networkID} argIdx := 2 if parentID != nil { query += ` AND p.parent_id = $` + string(rune('0'+argIdx)) args = append(args, *parentID) argIdx++ } else { query += ` AND p.parent_id IS NULL` } // Filter by visibility: include if network_all, inherited, OR user is a member query += ` AND (p.visibility = 'network_all' OR p.visibility = 'inherited' OR EXISTS (SELECT 1 FROM particle_members pm WHERE pm.particle_id = p.id AND pm.email = $` + string(rune('0'+argIdx)) + `))` args = append(args, requesterEmail) argIdx++ if len(filter.Types) > 0 { query += ` AND p.type = ANY($` + string(rune('0'+argIdx)) + `)` typeStrings := make([]string, len(filter.Types)) for i, t := range filter.Types { typeStrings[i] = string(t) } args = append(args, typeStrings) argIdx++ } if cursor != nil { if cursor.Direction == "before" { query += ` AND p.updated_at > $` + string(rune('0'+argIdx)) } else { query += ` AND p.updated_at < $` + string(rune('0'+argIdx)) } args = append(args, cursor.Position) argIdx++ } query += ` ORDER BY p.updated_at DESC LIMIT $` + string(rune('0'+argIdx)) args = append(args, limit) rows, err := r.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() return scanParticles(rows) } func scanParticles(rows pgx.Rows) ([]*Particle, error) { var particles []*Particle for rows.Next() { var p Particle if err := rows.Scan(&p.ID, &p.Type, &p.NetworkID, &p.ParentID, &p.CreatedByEmail, &p.Visibility, &p.Data, &p.UpdatedAt, &p.CreatedAt); err != nil { return nil, err } particles = append(particles, &p) } return particles, rows.Err() } func (r *repositoryImpl) setVisibility(ctx context.Context, id string, mode VisibilityMode) error { result, err := r.pool.Exec(ctx, `UPDATE particles SET visibility = $1, updated_at = NOW() WHERE id = $2`, mode, id, ) if err != nil { return err } if result.RowsAffected() == 0 { return errNotFound } return nil } func (r *repositoryImpl) addMembers(ctx context.Context, particleID string, emails []string) error { for _, email := range emails { _, err := r.pool.Exec(ctx, `INSERT INTO particle_members (particle_id, email) VALUES ($1, $2) ON CONFLICT (particle_id, email) DO NOTHING`, particleID, email, ) if err != nil { return err } } return nil } func (r *repositoryImpl) removeMembers(ctx context.Context, particleID string, emails []string) error { for _, email := range emails { _, err := r.pool.Exec(ctx, `DELETE FROM particle_members WHERE particle_id = $1 AND email = $2`, particleID, email, ) if err != nil { return err } } return nil } func (r *repositoryImpl) getMembers(ctx context.Context, particleID string) ([]string, error) { rows, err := r.pool.Query(ctx, `SELECT email FROM particle_members WHERE particle_id = $1`, particleID, ) if err != nil { return nil, err } defer rows.Close() var emails []string for rows.Next() { var email string if err := rows.Scan(&email); err != nil { return nil, err } emails = append(emails, email) } return emails, rows.Err() } func (r *repositoryImpl) getMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error) { if len(particleIDs) == 0 { return map[string][]string{}, nil } rows, err := r.pool.Query(ctx, `SELECT particle_id, email FROM particle_members WHERE particle_id = ANY($1)`, particleIDs, ) if err != nil { return nil, err } defer rows.Close() result := make(map[string][]string) for rows.Next() { var particleID, email string if err := rows.Scan(&particleID, &email); err != nil { return nil, err } result[particleID] = append(result[particleID], email) } return result, rows.Err() } func (r *repositoryImpl) getAncestorChain(ctx context.Context, particleID string) ([]*Particle, error) { rows, err := r.pool.Query(ctx, ` WITH RECURSIVE ancestors AS ( SELECT id, type, network_id, parent_id, created_by_email, visibility, data, updated_at, created_at FROM particles WHERE id = $1 UNION ALL SELECT p.id, p.type, p.network_id, p.parent_id, p.created_by_email, p.visibility, p.data, p.updated_at, p.created_at FROM particles p JOIN ancestors a ON p.id = a.parent_id ) SELECT id, type, network_id, parent_id, created_by_email, visibility, data, updated_at, created_at FROM ancestors; `, particleID) if err != nil { return nil, err } defer rows.Close() particles, err := scanParticles(rows) if err != nil { return nil, err } if len(particles) == 0 { return nil, errNotFound } return particles, nil } func (r *repositoryImpl) isMemberOf(ctx context.Context, particleID, email string) (bool, error) { var isMember bool err := r.pool.QueryRow(ctx, ` SELECT EXISTS(SELECT 1 FROM particle_members WHERE particle_id = $1 AND email = $2) `, particleID, email).Scan(&isMember) return isMember, err } func (r *repositoryImpl) markSeen(ctx context.Context, particleID, email string) error { _, err := r.pool.Exec(ctx, `INSERT INTO particle_seen (particle_id, email) VALUES ($1, $2) ON CONFLICT (particle_id, email) DO NOTHING`, particleID, email, ) return err } func (r *repositoryImpl) getSeenMap(ctx context.Context, particleIDs []string, email string) (map[string]bool, error) { if len(particleIDs) == 0 { return map[string]bool{}, nil } rows, err := r.pool.Query(ctx, `SELECT particle_id FROM particle_seen WHERE particle_id = ANY($1) AND email = $2`, particleIDs, email, ) if err != nil { return nil, err } defer rows.Close() result := make(map[string]bool) for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, err } result[id] = true } return result, rows.Err() } func (r *repositoryImpl) getUnseenCounts(ctx context.Context, streamIDs []string, email string) (map[string]int, error) { if len(streamIDs) == 0 { return map[string]int{}, nil } rows, err := r.pool.Query(ctx, ` SELECT p.parent_id, COUNT(*) FROM particles p WHERE p.parent_id = ANY($1) AND NOT EXISTS (SELECT 1 FROM particle_seen ps WHERE ps.particle_id = p.id AND ps.email = $2) AND (p.visibility = 'network_all' OR p.visibility = 'inherited' OR EXISTS (SELECT 1 FROM particle_members pm WHERE pm.particle_id = p.id AND pm.email = $2)) GROUP BY p.parent_id `, streamIDs, email) if err != nil { return nil, err } defer rows.Close() result := make(map[string]int) for rows.Next() { var parentID string var count int if err := rows.Scan(&parentID, &count); err != nil { return nil, err } result[parentID] = count } return result, rows.Err() } func (r *repositoryImpl) ack(ctx context.Context, particleID, email string) error { _, err := r.pool.Exec(ctx, `INSERT INTO particle_acks (particle_id, email) VALUES ($1, $2) ON CONFLICT (particle_id, email) DO NOTHING`, particleID, email, ) return err } func (r *repositoryImpl) getAcksMap(ctx context.Context, particleIDs []string) (map[string][]AckInfo, error) { if len(particleIDs) == 0 { return map[string][]AckInfo{}, nil } rows, err := r.pool.Query(ctx, `SELECT particle_id, email, acked_at FROM particle_acks WHERE particle_id = ANY($1) ORDER BY acked_at`, particleIDs, ) if err != nil { return nil, err } defer rows.Close() result := make(map[string][]AckInfo) for rows.Next() { var particleID string var info AckInfo if err := rows.Scan(&particleID, &info.Email, &info.AckedAt); err != nil { return nil, err } result[particleID] = append(result[particleID], info) } return result, rows.Err() }