Notice Board

CREATE EXTENSION IF NOT EXISTS "pgcrypto";

CREATE TABLE notices (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    org_id          UUID NOT NULL,
    title           TEXT NOT NULL,
    body            TEXT NOT NULL,
    state           TEXT NOT NULL CHECK (state IN ('draft', 'scheduled', 'published', 'archived')),
    audience_scope  TEXT NOT NULL CHECK (audience_scope IN ('all', 'custom')),
    created_by      UUID NOT NULL,
    scheduled_at    TIMESTAMPTZ,
    published_at    TIMESTAMPTZ,
    archived_at     TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT notices_id_org_unique UNIQUE (id, org_id),
    CONSTRAINT notices_schedule_state_chk CHECK (
        (state = 'draft' AND scheduled_at IS NULL AND published_at IS NULL AND archived_at IS NULL)
        OR (state = 'scheduled' AND scheduled_at IS NOT NULL AND published_at IS NULL AND archived_at IS NULL)
        OR (state = 'published' AND published_at IS NOT NULL AND archived_at IS NULL)
        OR (state = 'archived' AND published_at IS NOT NULL AND archived_at IS NOT NULL)
    )
);

CREATE INDEX notices_org_state_schedule_idx
    ON notices (org_id, state, scheduled_at, id);

CREATE INDEX notices_org_published_feed_idx
    ON notices (org_id, published_at DESC, id DESC)
    WHERE state = 'published';

CREATE TABLE notice_audience (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    org_id      UUID NOT NULL,
    notice_id   UUID NOT NULL,
    scope_type  TEXT NOT NULL CHECK (scope_type IN ('block', 'flat')),
    block_code  TEXT,
    flat_id     UUID,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    FOREIGN KEY (notice_id, org_id) REFERENCES notices(id, org_id) ON DELETE CASCADE,
    CONSTRAINT notice_audience_scope_chk CHECK (
        (scope_type = 'block' AND block_code IS NOT NULL AND flat_id IS NULL)
        OR (scope_type = 'flat' AND flat_id IS NOT NULL AND block_code IS NULL)
    )
);

CREATE UNIQUE INDEX notice_audience_uniqueness_idx
    ON notice_audience (
        notice_id,
        scope_type,
        COALESCE(block_code, ''),
        COALESCE(flat_id::text, '')
    );

CREATE INDEX notice_audience_org_notice_idx
    ON notice_audience (org_id, notice_id);

CREATE INDEX notice_audience_org_block_idx
    ON notice_audience (org_id, block_code)
    WHERE scope_type = 'block';

CREATE INDEX notice_audience_org_flat_idx
    ON notice_audience (org_id, flat_id)
    WHERE scope_type = 'flat';

DROP TABLE IF EXISTS notice_audience;
DROP TABLE IF EXISTS notices;

package notice

import (
    "context"
    "errors"
    "fmt"
    "strings"
    "time"

    "github.com/google/uuid"
)

type State string

type AudienceScope string

type AudienceType string

const (
    StateDraft     State = "draft"
    StateScheduled State = "scheduled"
    StatePublished State = "published"
    StateArchived  State = "archived"

    AudienceScopeAll    AudienceScope = "all"
    AudienceScopeCustom AudienceScope = "custom"

    AudienceTypeBlock AudienceType = "block"
    AudienceTypeFlat  AudienceType = "flat"
)

var (
    ErrInvalidSchedule       = errors.New("notice: scheduled notices require a future scheduled_at")
    ErrInvalidAudience       = errors.New("notice: invalid audience configuration")
    ErrNoticeNotFound        = errors.New("notice: not found")
    ErrCrossTenantRecipient  = errors.New("notice: recipient resolution crossed tenant boundary")
)

type Notice struct {
    ID            uuid.UUID
    OrgID         uuid.UUID
    Title         string
    Body          string
    State         State
    AudienceScope AudienceScope
    CreatedBy     uuid.UUID
    ScheduledAt   *time.Time
    PublishedAt   *time.Time
    ArchivedAt    *time.Time
    CreatedAt     time.Time
    UpdatedAt     time.Time
}

type AudienceRule struct {
    ScopeType AudienceType
    BlockCode string
    FlatID    *uuid.UUID
}

type CreateNoticeInput struct {
    Title         string
    Body          string
    PublishNow    bool
    ScheduledAt   *time.Time
    AudienceScope AudienceScope
    Audience      []AudienceRule
}

type Repository interface {
    Create(ctx context.Context, notice Notice, audience []AudienceRule) (Notice, error)
    ListVisibleResidentUserIDs(ctx context.Context, orgID, noticeID uuid.UUID) ([]uuid.UUID, error)
    UpdateState(ctx context.Context, orgID, noticeID uuid.UUID, from State, to State, publishedAt *time.Time) (Notice, error)
}

type Service struct {
    repo Repository
}

func NewService(repo Repository) *Service {
    return &Service{repo: repo}
}

func (s *Service) Create(ctx context.Context, orgID, actorID uuid.UUID, input CreateNoticeInput) (Notice, error) {
    title := strings.TrimSpace(input.Title)
    body := strings.TrimSpace(input.Body)
    if title == "" || body == "" {
        return Notice{}, fmt.Errorf("%w: title and body are required", ErrInvalidAudience)
    }

    state := StateDraft
    var publishedAt *time.Time

    if input.PublishNow {
        now := time.Now().UTC()
        state = StatePublished
        publishedAt = &now
        input.ScheduledAt = nil
    }

    if input.ScheduledAt != nil {
        if !input.ScheduledAt.After(time.Now().UTC()) {
            return Notice{}, ErrInvalidSchedule
        }
        state = StateScheduled
    }

    if err := validateAudience(input.AudienceScope, input.Audience); err != nil {
        return Notice{}, err
    }

    notice := Notice{
        OrgID:         orgID,
        Title:         title,
        Body:          body,
        State:         state,
        AudienceScope: input.AudienceScope,
        CreatedBy:     actorID,
        ScheduledAt:   input.ScheduledAt,
        PublishedAt:   publishedAt,
    }

    return s.repo.Create(ctx, notice, input.Audience)
}

func (s *Service) PreviewRecipients(ctx context.Context, orgID, noticeID uuid.UUID) ([]uuid.UUID, error) {
    recipients, err := s.repo.ListVisibleResidentUserIDs(ctx, orgID, noticeID)
    if err != nil {
        return nil, err
    }
    for _, userID := range recipients {
        if userID == uuid.Nil {
            return nil, ErrCrossTenantRecipient
        }
    }
    return recipients, nil
}

func (s *Service) Archive(ctx context.Context, orgID, noticeID uuid.UUID) (Notice, error) {
    publishedAt := time.Now().UTC()
    return s.repo.UpdateState(ctx, orgID, noticeID, StatePublished, StateArchived, &publishedAt)
}

func validateAudience(scope AudienceScope, audience []AudienceRule) error {
    switch scope {
    case AudienceScopeAll:
        if len(audience) != 0 {
            return fmt.Errorf("%w: all-scope notices must not persist audience rows", ErrInvalidAudience)
        }
        return nil
    case AudienceScopeCustom:
        if len(audience) == 0 {
            return fmt.Errorf("%w: custom-scope notices require at least one audience row", ErrInvalidAudience)
        }
        for _, rule := range audience {
            switch rule.ScopeType {
            case AudienceTypeBlock:
                if strings.TrimSpace(rule.BlockCode) == "" || rule.FlatID != nil {
                    return fmt.Errorf("%w: block audience requires block_code only", ErrInvalidAudience)
                }
            case AudienceTypeFlat:
                if rule.FlatID == nil || strings.TrimSpace(rule.BlockCode) != "" {
                    return fmt.Errorf("%w: flat audience requires flat_id only", ErrInvalidAudience)
                }
            default:
                return fmt.Errorf("%w: unsupported audience type %q", ErrInvalidAudience, rule.ScopeType)
            }
        }
        return nil
    default:
        return fmt.Errorf("%w: unsupported audience scope %q", ErrInvalidAudience, scope)
    }
}

package notice

import (
    "context"
    "time"

    "github.com/google/uuid"
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/rs/zerolog"
)

type PublisherStore interface {
    TryAcquireLeader(ctx context.Context) (bool, error)
    OrganizationsWithDueNotices(ctx context.Context, asOf time.Time, limit int) ([]uuid.UUID, error)
    PublishDueNotices(ctx context.Context, orgID uuid.UUID, asOf time.Time, limit int) ([]Notice, error)
}

type TimeSource interface {
    Now(ctx context.Context) (time.Time, error)
}

type DBTimeSource struct {
    pool *pgxpool.Pool
}

func NewDBTimeSource(pool *pgxpool.Pool) *DBTimeSource {
    return &DBTimeSource{pool: pool}
}

func (s *DBTimeSource) Now(ctx context.Context) (time.Time, error) {
    var current time.Time
    err := s.pool.QueryRow(ctx, "SELECT now()").Scan(&current)
    return current.UTC(), err
}

type Publisher struct {
    store      PublisherStore
    timeSource TimeSource
    logger     zerolog.Logger
    batchSize  int
    orgLimit   int
}

func NewPublisher(store PublisherStore, timeSource TimeSource, logger zerolog.Logger) *Publisher {
    return &Publisher{
        store:      store,
        timeSource: timeSource,
        logger:     logger,
        batchSize:  100,
        orgLimit:   25,
    }
}

func (p *Publisher) RunOnce(ctx context.Context) error {
    leader, err := p.store.TryAcquireLeader(ctx)
    if err != nil {
        return err
    }
    if !leader {
        p.logger.Info().Ctx(ctx).Msg("notice publisher skipped because another worker owns the lock")
        return nil
    }

    asOf, err := p.timeSource.Now(ctx)
    if err != nil {
        return err
    }

    orgIDs, err := p.store.OrganizationsWithDueNotices(ctx, asOf, p.orgLimit)
    if err != nil {
        return err
    }

    for _, orgID := range orgIDs {
        published, err := p.store.PublishDueNotices(ctx, orgID, asOf, p.batchSize)
        if err != nil {
            return err
        }
        for _, notice := range published {
            p.logger.Info().Ctx(ctx).
                Str("org_id", notice.OrgID.String()).
                Str("notice_id", notice.ID.String()).
                Interface("scheduled_at", notice.ScheduledAt).
                Interface("published_at", notice.PublishedAt).
                Msg("notice published")
        }
    }

    return nil
}

func PublishDueSQL() string {
    return "WITH due AS (" +
        " SELECT id" +
        " FROM notices" +
        " WHERE org_id = $1 AND state = 'scheduled' AND scheduled_at <= $2" +
        " ORDER BY scheduled_at ASC, id ASC" +
        " LIMIT $3" +
        " FOR UPDATE SKIP LOCKED" +
        ")" +
        " UPDATE notices AS n" +
        " SET state = 'published', published_at = $2, updated_at = $2" +
        " FROM due" +
        " WHERE n.id = due.id AND n.org_id = $1" +
        " RETURNING n.id, n.org_id, n.title, n.body, n.state, n.audience_scope, n.created_by, n.scheduled_at, n.published_at, n.archived_at, n.created_at, n.updated_at"
}