diff --git a/cmd/deliberi/main.go b/cmd/deliberi/main.go
new file mode 100644
index 00000000..4760b66d
--- /dev/null
+++ b/cmd/deliberi/main.go
@@ -0,0 +1,55 @@
+package main
+
+import (
+ "context"
+ "log/slog"
+ "os"
+ "os/signal"
+ "syscall"
+
+ "github.com/carlmjohnson/versioninfo"
+ "github.com/urfave/cli/v3"
+ "tangled.org/core/deliberi"
+ "tangled.org/core/deliberi/config"
+ "tangled.org/core/log"
+)
+
+func main() {
+ if err := run(os.Args); err != nil {
+ slog.Error("error running deliberi", "err", err)
+ os.Exit(-1)
+ }
+}
+
+func run(args []string) error {
+ ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+ defer cancel()
+
+ logger := log.New("deliberi")
+ slog.SetDefault(logger)
+ ctx = log.IntoContext(ctx, logger)
+
+ app := cli.Command{
+ Name: "deliberi",
+ Usage: "tngl.sh notification + email service",
+ Version: versioninfo.Short(),
+ }
+ app.Commands = []*cli.Command{
+ {
+ Name: "serve",
+ Usage: "run the deliberi daemon",
+ Action: runDeliberi,
+ },
+ }
+ return app.Run(ctx, args)
+}
+
+func runDeliberi(ctx context.Context, cmd *cli.Command) error {
+ logger := log.FromContext(ctx)
+ cfg, err := config.Load(ctx)
+ if err != nil {
+ return err
+ }
+ logger.Debug("config loaded", "config", cfg)
+ return deliberi.Run(ctx, cfg)
+}
diff --git a/deliberi/bobbin.go b/deliberi/bobbin.go
new file mode 100644
index 00000000..ccb06554
--- /dev/null
+++ b/deliberi/bobbin.go
@@ -0,0 +1,71 @@
+package deliberi
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ indigoxrpc "github.com/bluesky-social/indigo/xrpc"
+ "tangled.org/core/api/tangled"
+)
+
+type recipientResolver interface {
+ ListRecipients(ctx context.Context, uri string) ([]string, error)
+ RepoOwner(ctx context.Context, repoDid string) (ownerDid, name string, err error)
+}
+
+type bobbinClient struct {
+ xc *indigoxrpc.Client
+}
+
+func newBobbinClient(apiUrl string) *bobbinClient {
+ return &bobbinClient{
+ xc: &indigoxrpc.Client{
+ Host: strings.TrimRight(apiUrl, "/"),
+ Client: &http.Client{Timeout: 10 * time.Second},
+ },
+ }
+}
+
+func (c *bobbinClient) ListRecipients(ctx context.Context, uri string) ([]string, error) {
+ out, err := tangled.TempNotificationListRecipients(ctx, c.xc, uri)
+ if err != nil {
+ return nil, fmt.Errorf("calling %s: %w", tangled.TempNotificationListRecipientsNSID, err)
+ }
+ return out.Dids, nil
+}
+
+// repoByRepoDid mirrors sh.tangled.repo.getRepoByRepoDid's output, but keeps the
+// record raw: the owner comes from the uri's authority, so a record we can't
+// type-decode still resolves an owner.
+type repoByRepoDid struct {
+ Uri string `json:"uri"`
+ Value json.RawMessage `json:"value"`
+}
+
+// RepoOwner resolves a repo DID to the did that holds its sh.tangled.repo
+// record, along with the repo's cosmetic name.
+func (c *bobbinClient) RepoOwner(ctx context.Context, repoDid string) (string, string, error) {
+ var out repoByRepoDid
+ params := map[string]any{"repoDid": repoDid}
+ if err := c.xc.Do(ctx, indigoxrpc.Query, "", tangled.RepoGetRepoByRepoDidNSID, params, nil, &out); err != nil {
+ return "", "", fmt.Errorf("calling %s: %w", tangled.RepoGetRepoByRepoDidNSID, err)
+ }
+
+ owner := syntax.ATURI(out.Uri).Authority().String()
+ if owner == "" {
+ return "", "", fmt.Errorf("no authority in uri %q", out.Uri)
+ }
+
+ // the name is cosmetic, so a decode failure is not fatal.
+ var rec struct {
+ Name string `json:"name"`
+ }
+ _ = json.Unmarshal(out.Value, &rec)
+
+ return owner, rec.Name, nil
+}
diff --git a/deliberi/config/config.go b/deliberi/config/config.go
new file mode 100644
index 00000000..09753b84
--- /dev/null
+++ b/deliberi/config/config.go
@@ -0,0 +1,57 @@
+package config
+
+import (
+ "context"
+
+ "github.com/sethvargo/go-envconfig"
+)
+
+type Config struct {
+ // ListenAddr is where deliberi's xrpc + health server binds.
+ ListenAddr string `env:"DELIBERI_LISTEN_ADDR, default=0.0.0.0:6565"`
+
+ // Hostname is deliberi's public hostname; it derives deliberi's did:web,
+ // used as the audience for verifying inbound service-auth tokens.
+ Hostname string `env:"DELIBERI_HOSTNAME, required"`
+
+ DbPath string `env:"DELIBERI_DB_PATH, default=deliberi.db"`
+
+ PlcUrl string `env:"DELIBERI_PLC_URL, default=https://plc.directory"`
+ JetstreamEndpoint string `env:"DELIBERI_JETSTREAM_ENDPOINT, default=wss://jetstream1.us-east.bsky.network/subscribe"`
+
+ // BobbinApiUrl hosts listRecipients (subscriber fan-out); called as a
+ // plain internal xrpc, no auth.
+ BobbinApiUrl string `env:"DELIBERI_BOBBIN_API_URL, default=https://api.tangled.org"`
+
+ // BaseURL is the public frontend URL used to build links in digest emails.
+ BaseURL string `env:"DELIBERI_BASE_URL, default=https://tangled.org"`
+
+ Pds PdsConfig `env:",prefix=DELIBERI_PDS_"`
+ Resend ResendConfig `env:",prefix=DELIBERI_RESEND_"`
+
+ Dev bool `env:"DELIBERI_DEV, default=false"`
+}
+
+type PdsConfig struct {
+ Host string `env:"HOST, default=https://tngl.sh"`
+ UserDomain string `env:"USER_DOMAIN, default=.tngl.sh"`
+ AdminSecret string `env:"ADMIN_SECRET"`
+}
+
+type ResendConfig struct {
+ ApiKey string `env:"API_KEY"`
+ SentFrom string `env:"SENT_FROM, default=noreply@notifs.tangled.sh"`
+ AssetsURL string `env:"ASSETS_URL, default=https://assets.tangled.network/email/"`
+}
+
+func (c *Config) SignupEnabled() bool {
+ return c.Pds.AdminSecret != ""
+}
+
+func Load(ctx context.Context) (*Config, error) {
+ var cfg Config
+ if err := envconfig.Process(ctx, &cfg); err != nil {
+ return nil, err
+ }
+ return &cfg, nil
+}
diff --git a/deliberi/db/cache.go b/deliberi/db/cache.go
new file mode 100644
index 00000000..2f19c65a
--- /dev/null
+++ b/deliberi/db/cache.go
@@ -0,0 +1,37 @@
+package db
+
+func PutRepoName(e Execer, repoDid, ownerDid, name string) error {
+ _, err := e.Exec(
+ `insert into repo_names (repo_did, name, owner_did) values (?, ?, ?)
+ on conflict(repo_did) do update set name = excluded.name, owner_did = excluded.owner_did`,
+ repoDid, name, ownerDid,
+ )
+ return err
+}
+
+func GetRepoName(e Execer, repoDid string) string {
+ var name string
+ _ = e.QueryRow(`select name from repo_names where repo_did = ?`, repoDid).Scan(&name)
+ return name
+}
+
+func GetRepoOwner(e Execer, repoDid string) string {
+ var owner string
+ _ = e.QueryRow(`select owner_did from repo_names where repo_did = ?`, repoDid).Scan(&owner)
+ return owner
+}
+
+func PutEntityTitle(e Execer, atUri, title string) error {
+ _, err := e.Exec(
+ `insert into entity_titles (at_uri, title) values (?, ?)
+ on conflict(at_uri) do update set title = excluded.title`,
+ atUri, title,
+ )
+ return err
+}
+
+func GetEntityTitle(e Execer, atUri string) string {
+ var title string
+ _ = e.QueryRow(`select title from entity_titles where at_uri = ?`, atUri).Scan(&title)
+ return title
+}
diff --git a/deliberi/db/cursor.go b/deliberi/db/cursor.go
new file mode 100644
index 00000000..bd30fb40
--- /dev/null
+++ b/deliberi/db/cursor.go
@@ -0,0 +1,20 @@
+package db
+
+import "fmt"
+
+func (d *DB) GetLastTimeUs() (int64, error) {
+ var t int64
+ if err := d.QueryRow(`select last_time_us from jetstream_cursor where id = 0`).Scan(&t); err != nil {
+ return 0, fmt.Errorf("no saved cursor: %w", err)
+ }
+ return t, nil
+}
+
+func (d *DB) SaveLastTimeUs(t int64) error {
+ _, err := d.Exec(
+ `insert into jetstream_cursor (id, last_time_us) values (0, ?)
+ on conflict(id) do update set last_time_us = excluded.last_time_us`,
+ t,
+ )
+ return err
+}
diff --git a/deliberi/db/db.go b/deliberi/db/db.go
new file mode 100644
index 00000000..51acbc6f
--- /dev/null
+++ b/deliberi/db/db.go
@@ -0,0 +1,154 @@
+package db
+
+import (
+ "context"
+ "database/sql"
+ "log/slog"
+ "strings"
+
+ _ "github.com/mattn/go-sqlite3"
+ "tangled.org/core/log"
+ "tangled.org/core/orm"
+)
+
+type DB struct {
+ *sql.DB
+ logger *slog.Logger
+}
+
+type Execer interface {
+ Query(query string, args ...any) (*sql.Rows, error)
+ QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
+ QueryRow(query string, args ...any) *sql.Row
+ Exec(query string, args ...any) (sql.Result, error)
+}
+
+func Make(ctx context.Context, dbPath string) (*DB, error) {
+ opts := []string{
+ "_foreign_keys=1",
+ "_journal_mode=WAL",
+ "_synchronous=NORMAL",
+ "_busy_timeout=5000",
+ }
+
+ logger := log.SubLogger(log.FromContext(ctx), "db")
+
+ db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&"))
+ if err != nil {
+ return nil, err
+ }
+
+ conn, err := db.Conn(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer conn.Close()
+
+ _, err = conn.ExecContext(ctx, schema)
+ if err != nil {
+ return nil, err
+ }
+
+ if err := runMigrations(conn, logger); err != nil {
+ return nil, err
+ }
+
+ return &DB{db, logger}, nil
+}
+
+func (d *DB) Close() error {
+ return d.DB.Close()
+}
+
+const schema = `
+create table if not exists emails (
+ id integer primary key autoincrement,
+ did text not null,
+ email text not null,
+ verified integer not null default 0,
+ verification_code text not null,
+ last_sent text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+ is_primary integer not null default 0,
+ created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+ unique(did, email)
+);
+
+create table if not exists signups_inflight (
+ id integer primary key autoincrement,
+ email text not null unique,
+ invite_code text not null,
+ created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
+);
+
+-- one materialized row per (recipient, source record). deliberi owns these;
+-- read/emailed are inline. unique(recipient_did, at_uri) dedupes fan-out.
+create table if not exists notifications (
+ id integer primary key autoincrement,
+ recipient_did text not null,
+ at_uri text not null,
+ type text not null,
+ actor_did text not null,
+ repo_did text not null default '',
+ entity_at text not null default '',
+ entity_title text not null default '',
+ read integer not null default 0,
+ emailed integer not null default 0,
+ created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+ unique(recipient_did, at_uri)
+);
+create index if not exists idx_deliberi_notifs_recipient on notifications(recipient_did, read, created);
+create index if not exists idx_deliberi_notifs_digest on notifications(recipient_did, emailed, read, created);
+
+-- small denormalization caches the ingester fills from repo/issue/pull
+-- records so notifications render human names without extra lookups.
+create table if not exists repo_names (
+ repo_did text primary key,
+ name text not null,
+ owner_did text not null default ''
+);
+create table if not exists entity_titles (
+ at_uri text primary key,
+ title text not null
+);
+
+create table if not exists jetstream_cursor (
+ id integer primary key check (id = 0),
+ last_time_us integer not null
+);
+
+create table if not exists migrations (
+ name text primary key
+);
+
+create table if not exists notification_preferences (
+ id integer primary key autoincrement,
+ user_did text not null unique,
+ repo_starred integer not null default 1,
+ issue_created integer not null default 1,
+ issue_commented integer not null default 1,
+ pull_created integer not null default 1,
+ pull_commented integer not null default 1,
+ followed integer not null default 1,
+ pull_merged integer not null default 1,
+ issue_closed integer not null default 1,
+ user_mentioned integer not null default 1,
+ email_notifications integer not null default 0
+);
+`
+
+func runMigrations(conn *sql.Conn, logger *slog.Logger) error {
+ if err := orm.RunMigration(conn, logger, "add-owner-did-to-repo-names", func(tx *sql.Tx) error {
+ _, err := tx.Exec(`alter table repo_names add column owner_did text not null default ''`)
+ if err != nil && !isColumnExistsErr(err) {
+ return err
+ }
+ return nil
+ }); err != nil {
+ return err
+ }
+ return nil
+}
+
+func isColumnExistsErr(err error) bool {
+ return err != nil && strings.Contains(err.Error(), "duplicate column name")
+}
diff --git a/deliberi/db/email.go b/deliberi/db/email.go
new file mode 100644
index 00000000..c3d98311
--- /dev/null
+++ b/deliberi/db/email.go
@@ -0,0 +1,295 @@
+package db
+
+import (
+ "strings"
+ "time"
+
+ "tangled.org/core/deliberi/models"
+)
+
+func GetPrimaryEmail(e Execer, did string) (models.Email, error) {
+ query := `
+ select id, did, email, verified, is_primary, verification_code, last_sent, created
+ from emails
+ where did = ? and is_primary = true
+ `
+ var email models.Email
+ var createdStr string
+ var lastSent string
+ err := e.QueryRow(query, did).Scan(&email.ID, &email.Did, &email.Address, &email.Verified, &email.Primary, &email.VerificationCode, &lastSent, &createdStr)
+ if err != nil {
+ return models.Email{}, err
+ }
+ email.CreatedAt, err = time.Parse(time.RFC3339, createdStr)
+ if err != nil {
+ return models.Email{}, err
+ }
+ parsedTime, err := time.Parse(time.RFC3339, lastSent)
+ if err != nil {
+ return models.Email{}, err
+ }
+ email.LastSent = &parsedTime
+ return email, nil
+}
+
+func GetEmail(e Execer, did string, em string) (models.Email, error) {
+ query := `
+ select id, did, email, verified, is_primary, verification_code, last_sent, created
+ from emails
+ where did = ? and email = ?
+ `
+ var email models.Email
+ var createdStr string
+ var lastSent string
+ err := e.QueryRow(query, did, em).Scan(&email.ID, &email.Did, &email.Address, &email.Verified, &email.Primary, &email.VerificationCode, &lastSent, &createdStr)
+ if err != nil {
+ return models.Email{}, err
+ }
+ email.CreatedAt, err = time.Parse(time.RFC3339, createdStr)
+ if err != nil {
+ return models.Email{}, err
+ }
+ parsedTime, err := time.Parse(time.RFC3339, lastSent)
+ if err != nil {
+ return models.Email{}, err
+ }
+ email.LastSent = &parsedTime
+ return email, nil
+}
+
+func GetDidForEmail(e Execer, em string) (string, error) {
+ query := `
+ select did
+ from emails
+ where email = ?
+ `
+ var did string
+ err := e.QueryRow(query, em).Scan(&did)
+ if err != nil {
+ return "", err
+ }
+ return did, nil
+}
+
+// GetEmailToDid maps committer emails to dids, optionally restricting to
+// verified emails. did-prefixed inputs pass through as already-resolved. this
+// is what resolves git committer emails to tangled accounts.
+func GetEmailToDid(e Execer, emails []string, isVerifiedFilter bool) (map[string]string, error) {
+ if len(emails) == 0 {
+ return make(map[string]string), nil
+ }
+
+ verifiedFilter := 0
+ if isVerifiedFilter {
+ verifiedFilter = 1
+ }
+
+ assoc := make(map[string]string)
+
+ placeholders := make([]string, 0, len(emails))
+ args := make([]any, 1, len(emails)+1)
+
+ args[0] = verifiedFilter
+ for _, email := range emails {
+ if strings.HasPrefix(email, "did:") {
+ assoc[email] = email
+ continue
+ }
+ placeholders = append(placeholders, "?")
+ args = append(args, email)
+ }
+
+ if len(placeholders) == 0 {
+ return assoc, nil
+ }
+
+ query := `
+ select email, did
+ from emails
+ where
+ verified = ?
+ and email in (` + strings.Join(placeholders, ",") + `)
+ `
+
+ rows, err := e.Query(query, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ for rows.Next() {
+ var email, did string
+ if err := rows.Scan(&email, &did); err != nil {
+ return nil, err
+ }
+ assoc[email] = did
+ }
+
+ return assoc, rows.Err()
+}
+
+func GetVerificationCodeForEmail(e Execer, did string, email string) (string, error) {
+ query := `
+ select verification_code
+ from emails
+ where did = ? and email = ?
+ `
+ var code string
+ err := e.QueryRow(query, did, email).Scan(&code)
+ if err != nil {
+ return "", err
+ }
+ return code, nil
+}
+
+func CheckEmailExists(e Execer, did string, email string) (bool, error) {
+ query := `
+ select count(*)
+ from emails
+ where did = ? and email = ?
+ `
+ var count int
+ err := e.QueryRow(query, did, email).Scan(&count)
+ if err != nil {
+ return false, err
+ }
+ return count > 0, nil
+}
+
+func CheckEmailExistsAtAll(e Execer, email string) (bool, error) {
+ query := `
+ select count(*)
+ from emails
+ where email = ?
+ `
+ var count int
+ err := e.QueryRow(query, email).Scan(&count)
+ if err != nil {
+ return false, err
+ }
+ return count > 0, nil
+}
+
+func CheckValidVerificationCode(e Execer, did string, email string, code string) (bool, error) {
+ query := `
+ select count(*)
+ from emails
+ where did = ? and email = ? and verification_code = ?
+ `
+ var count int
+ err := e.QueryRow(query, did, email, code).Scan(&count)
+ if err != nil {
+ return false, err
+ }
+ return count > 0, nil
+}
+
+func AddEmail(e Execer, email models.Email) error {
+ countQuery := `
+ select count(*)
+ from emails
+ where did = ?
+ `
+ var count int
+ err := e.QueryRow(countQuery, email.Did).Scan(&count)
+ if err != nil {
+ return err
+ }
+
+ // first email for a did becomes primary
+ if count == 0 {
+ email.Primary = true
+ }
+
+ query := `
+ insert into emails (did, email, verified, is_primary, verification_code)
+ values (?, ?, ?, ?, ?)
+ `
+ _, err = e.Exec(query, email.Did, email.Address, email.Verified, email.Primary, email.VerificationCode)
+ return err
+}
+
+func DeleteEmail(e Execer, did string, email string) error {
+ query := `
+ delete from emails
+ where did = ? and email = ?
+ `
+ _, err := e.Exec(query, did, email)
+ return err
+}
+
+func MarkEmailVerified(e Execer, did string, email string) error {
+ query := `
+ update emails
+ set verified = true
+ where did = ? and email = ?
+ `
+ _, err := e.Exec(query, did, email)
+ return err
+}
+
+func MakeEmailPrimary(e Execer, did string, email string) error {
+ query1 := `
+ update emails
+ set is_primary = false
+ where did = ?
+ `
+ _, err := e.Exec(query1, did)
+ if err != nil {
+ return err
+ }
+
+ query2 := `
+ update emails
+ set is_primary = true
+ where did = ? and email = ?
+ `
+ _, err = e.Exec(query2, did, email)
+ return err
+}
+
+func GetAllEmails(e Execer, did string) ([]models.Email, error) {
+ query := `
+ select did, email, verified, is_primary, verification_code, last_sent, created
+ from emails
+ where did = ?
+ `
+ rows, err := e.Query(query, did)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ var emails []models.Email
+ for rows.Next() {
+ var email models.Email
+ var createdStr string
+ var lastSent string
+ err := rows.Scan(&email.Did, &email.Address, &email.Verified, &email.Primary, &email.VerificationCode, &lastSent, &createdStr)
+ if err != nil {
+ return nil, err
+ }
+ email.CreatedAt, err = time.Parse(time.RFC3339, createdStr)
+ if err != nil {
+ return nil, err
+ }
+ parsedTime, err := time.Parse(time.RFC3339, lastSent)
+ if err != nil {
+ return nil, err
+ }
+ email.LastSent = &parsedTime
+ emails = append(emails, email)
+ }
+ return emails, nil
+}
+
+func UpdateVerificationCode(e Execer, did string, email string, code string) error {
+ query := `
+ update emails
+ set verification_code = ?,
+ last_sent = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
+ where did = ? and email = ?
+ `
+ _, err := e.Exec(query, code, did, email)
+ return err
+}
diff --git a/deliberi/db/notifications.go b/deliberi/db/notifications.go
new file mode 100644
index 00000000..ea1dd4f0
--- /dev/null
+++ b/deliberi/db/notifications.go
@@ -0,0 +1,233 @@
+package db
+
+import (
+ "database/sql"
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "tangled.org/core/deliberi/models"
+ "tangled.org/core/orm"
+)
+
+// CreateNotification inserts a row, deduped on (recipient_did, at_uri).
+func CreateNotification(e Execer, n *models.Notification) error {
+ query := `
+ insert into notifications
+ (recipient_did, at_uri, type, actor_did, repo_did, entity_at, entity_title, read)
+ values (?, ?, ?, ?, ?, ?, ?, ?)
+ on conflict(recipient_did, at_uri) do nothing
+ `
+ res, err := e.Exec(query,
+ n.RecipientDid, n.AtUri, string(n.Type), n.ActorDid,
+ n.RepoDid, n.EntityAt, n.EntityTitle, n.Read,
+ )
+ if err != nil {
+ return fmt.Errorf("failed to create notification: %w", err)
+ }
+ if id, err := res.LastInsertId(); err == nil {
+ n.ID = id
+ }
+ return nil
+}
+
+const notifCols = `id, recipient_did, at_uri, type, actor_did, repo_did, entity_at, entity_title, read, emailed, created`
+
+func scanNotification(rows interface{ Scan(...any) error }) (*models.Notification, error) {
+ var n models.Notification
+ var typeStr, createdStr string
+ if err := rows.Scan(
+ &n.ID, &n.RecipientDid, &n.AtUri, &typeStr, &n.ActorDid,
+ &n.RepoDid, &n.EntityAt, &n.EntityTitle, &n.Read, &n.Emailed, &createdStr,
+ ); err != nil {
+ return nil, err
+ }
+ n.Type = models.NotificationType(typeStr)
+ n.Created, _ = time.Parse(time.RFC3339, createdStr)
+ return &n, nil
+}
+
+func GetNotifications(e Execer, recipientDid string, limit int, filters ...orm.Filter) ([]*models.Notification, error) {
+ conds := []string{"recipient_did = ?"}
+ args := []any{recipientDid}
+ for _, f := range filters {
+ conds = append(conds, f.Condition())
+ args = append(args, f.Arg()...)
+ }
+ query := fmt.Sprintf("select %s from notifications where %s order by created desc", notifCols, strings.Join(conds, " and "))
+ if limit > 0 {
+ query += fmt.Sprintf(" limit %d", limit)
+ }
+ rows, err := e.Query(query, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []*models.Notification
+ for rows.Next() {
+ n, err := scanNotification(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, n)
+ }
+ return out, rows.Err()
+}
+
+func CountNotifications(e Execer, recipientDid string, filters ...orm.Filter) (int64, error) {
+ conds := []string{"recipient_did = ?"}
+ args := []any{recipientDid}
+ for _, f := range filters {
+ conds = append(conds, f.Condition())
+ args = append(args, f.Arg()...)
+ }
+ query := fmt.Sprintf("select count(*) from notifications where %s", strings.Join(conds, " and "))
+ var count int64
+ if err := e.QueryRow(query, args...).Scan(&count); err != nil {
+ return 0, err
+ }
+ return count, nil
+}
+
+func MarkRead(e Execer, recipientDid, atUri string, read bool) error {
+ _, err := e.Exec(`update notifications set read = ? where recipient_did = ? and at_uri = ?`, read, recipientDid, atUri)
+ return err
+}
+
+func MarkAllRead(e Execer, recipientDid string) error {
+ _, err := e.Exec(`update notifications set read = 1 where recipient_did = ? and read = 0`, recipientDid)
+ return err
+}
+
+func MarkEmailed(e Execer, ids []int64) error {
+ if len(ids) == 0 {
+ return nil
+ }
+ ph := make([]string, len(ids))
+ args := make([]any, len(ids))
+ for i, id := range ids {
+ ph[i] = "?"
+ args[i] = id
+ }
+ _, err := e.Exec(fmt.Sprintf(`update notifications set emailed = 1 where id in (%s)`, strings.Join(ph, ", ")), args...)
+ return err
+}
+
+func GetPendingEmailDigestRecipients(e Execer, olderThan time.Time) ([]string, error) {
+ ph := make([]string, len(models.EmailNotificationTypes))
+ args := []any{olderThan.UTC().Format(time.RFC3339)}
+ for i, t := range models.EmailNotificationTypes {
+ ph[i] = "?"
+ args = append(args, string(t))
+ }
+ query := fmt.Sprintf(`
+ select distinct n.recipient_did
+ from notifications n
+ join notification_preferences np on np.user_did = n.recipient_did
+ join emails em on em.did = n.recipient_did and em.is_primary = 1 and em.verified = 1
+ where n.emailed = 0 and n.read = 0 and n.created < ?
+ and np.email_notifications = 1
+ and n.type in (%s)
+ `, strings.Join(ph, ", "))
+ rows, err := e.Query(query, args...)
+ if err != nil {
+ return nil, fmt.Errorf("failed to query digest recipients: %w", err)
+ }
+ defer rows.Close()
+ var dids []string
+ for rows.Next() {
+ var did string
+ if err := rows.Scan(&did); err != nil {
+ return nil, err
+ }
+ dids = append(dids, did)
+ }
+ return dids, rows.Err()
+}
+
+func GetPendingNotificationsForEmailDigest(e Execer, recipientDid string, olderThan time.Time) ([]*models.Notification, error) {
+ ph := make([]string, len(models.EmailNotificationTypes))
+ args := []any{recipientDid, olderThan.UTC().Format(time.RFC3339)}
+ for i, t := range models.EmailNotificationTypes {
+ ph[i] = "?"
+ args = append(args, string(t))
+ }
+ query := fmt.Sprintf(`
+ select %s from notifications
+ where recipient_did = ? and emailed = 0 and read = 0 and created < ?
+ and type in (%s)
+ order by created desc
+ `, notifCols, strings.Join(ph, ", "))
+ rows, err := e.Query(query, args...)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []*models.Notification
+ for rows.Next() {
+ n, err := scanNotification(rows)
+ if err != nil {
+ return nil, err
+ }
+ // repo names are not stored on the row: resolve them here, where the
+ // digest needs them, so a rename shows up in the next email.
+ n.RepoName = GetRepoName(e, n.RepoDid)
+ out = append(out, n)
+ }
+ return out, rows.Err()
+}
+
+func GetNotificationPreference(e Execer, userDid string) (*models.NotificationPreferences, error) {
+ query := `
+ select id, user_did, repo_starred, issue_created, issue_commented, pull_created,
+ pull_commented, followed, pull_merged, issue_closed, user_mentioned, email_notifications
+ from notification_preferences
+ where user_did = ?
+ `
+ var p models.NotificationPreferences
+ var userDidStr string
+ err := e.QueryRow(query, userDid).Scan(
+ &p.ID, &userDidStr, &p.RepoStarred, &p.IssueCreated, &p.IssueCommented, &p.PullCreated,
+ &p.PullCommented, &p.Followed, &p.PullMerged, &p.IssueClosed, &p.UserMentioned, &p.EmailNotifications,
+ )
+ if errors.Is(err, sql.ErrNoRows) {
+ // no row yet: defaults so reads never fail for an uncustomized user
+ return models.DefaultNotificationPreferences(syntax.DID(userDid)), nil
+ }
+ if err != nil {
+ // a real failure must not read as "user wants everything": callers
+ // deliver on these prefs, so opting out has to survive a db error.
+ return nil, fmt.Errorf("failed to query notification preferences: %w", err)
+ }
+ p.UserDid = syntax.DID(userDidStr)
+ return &p, nil
+}
+
+func UpsertNotificationPreferences(e Execer, prefs *models.NotificationPreferences) error {
+ query := `
+ insert into notification_preferences
+ (user_did, repo_starred, issue_created, issue_commented, pull_created,
+ pull_commented, followed, pull_merged, issue_closed, user_mentioned, email_notifications)
+ values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ on conflict(user_did) do update set
+ repo_starred = excluded.repo_starred,
+ issue_created = excluded.issue_created,
+ issue_commented = excluded.issue_commented,
+ pull_created = excluded.pull_created,
+ pull_commented = excluded.pull_commented,
+ followed = excluded.followed,
+ pull_merged = excluded.pull_merged,
+ issue_closed = excluded.issue_closed,
+ user_mentioned = excluded.user_mentioned,
+ email_notifications = excluded.email_notifications
+ `
+ _, err := e.Exec(query,
+ prefs.UserDid.String(),
+ prefs.RepoStarred, prefs.IssueCreated, prefs.IssueCommented, prefs.PullCreated,
+ prefs.PullCommented, prefs.Followed, prefs.PullMerged, prefs.IssueClosed,
+ prefs.UserMentioned, prefs.EmailNotifications,
+ )
+ return err
+}
diff --git a/deliberi/db/notifications_test.go b/deliberi/db/notifications_test.go
new file mode 100644
index 00000000..74dd06a7
--- /dev/null
+++ b/deliberi/db/notifications_test.go
@@ -0,0 +1,75 @@
+package db
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "tangled.org/core/deliberi/models"
+ "tangled.org/core/orm"
+)
+
+func testDB(t *testing.T) *DB {
+ t.Helper()
+ d, err := Make(context.Background(), filepath.Join(t.TempDir(), "x.db"))
+ if err != nil {
+ t.Fatalf("Make: %v", err)
+ }
+ t.Cleanup(func() { d.Close() })
+ return d
+}
+
+// one source record fans out to a row per recipient; read state is per row
+// (recipient_did, at_uri), so marking one recipient's row read leaves another
+// recipient's untouched.
+func TestNotificationsIsolatedPerRecipient(t *testing.T) {
+ d := testDB(t)
+ const uri = "at://did:plc:author/sh.tangled.feed.comment/abc"
+
+ for _, did := range []string{"did:plc:alice", "did:plc:bob"} {
+ if err := CreateNotification(d, &models.Notification{
+ RecipientDid: did,
+ AtUri: uri,
+ Type: models.NotificationTypeIssueCommented,
+ ActorDid: "did:plc:author",
+ }); err != nil {
+ t.Fatalf("CreateNotification %s: %v", did, err)
+ }
+ }
+
+ if err := MarkRead(d, "did:plc:alice", uri, true); err != nil {
+ t.Fatalf("MarkRead: %v", err)
+ }
+
+ aliceUnread, _ := CountNotifications(d, "did:plc:alice", orm.FilterEq("read", 0))
+ bobUnread, _ := CountNotifications(d, "did:plc:bob", orm.FilterEq("read", 0))
+ if aliceUnread != 0 {
+ t.Fatalf("alice unread = %d, want 0", aliceUnread)
+ }
+ if bobUnread != 1 {
+ t.Fatalf("bob unread = %d, want 1 (must not inherit alice's read)", bobUnread)
+ }
+}
+
+// CreateNotification dedupes on (recipient_did, at_uri): a firehose replay or a
+// user both subscribed and mentioned yields a single row.
+func TestCreateNotificationDedupe(t *testing.T) {
+ d := testDB(t)
+ const uri = "at://did:plc:repo/sh.tangled.repo.issue/abc"
+
+ for range 2 {
+ if err := CreateNotification(d, &models.Notification{
+ RecipientDid: "did:plc:alice",
+ AtUri: uri,
+ Type: models.NotificationTypeIssueCreated,
+ ActorDid: "did:plc:author",
+ }); err != nil {
+ t.Fatalf("CreateNotification: %v", err)
+ }
+ }
+
+ count, _ := CountNotifications(d, "did:plc:alice")
+ if count != 1 {
+ t.Fatalf("row count = %d, want 1 (deduped)", count)
+ }
+}
diff --git a/deliberi/db/signup.go b/deliberi/db/signup.go
new file mode 100644
index 00000000..f4fce644
--- /dev/null
+++ b/deliberi/db/signup.go
@@ -0,0 +1,24 @@
+package db
+
+import (
+ "tangled.org/core/deliberi/models"
+)
+
+func AddInflightSignup(e Execer, signup models.InflightSignup) error {
+ query := `insert or replace into signups_inflight (email, invite_code) values (?, ?)`
+ _, err := e.Exec(query, signup.Email, signup.InviteCode)
+ return err
+}
+
+func DeleteInflightSignup(e Execer, email string) error {
+ query := `delete from signups_inflight where email = ?`
+ _, err := e.Exec(query, email)
+ return err
+}
+
+func GetEmailForCode(e Execer, inviteCode string) (string, error) {
+ query := `select email from signups_inflight where invite_code = ?`
+ var email string
+ err := e.QueryRow(query, inviteCode).Scan(&email)
+ return email, err
+}
diff --git a/deliberi/deliberi.go b/deliberi/deliberi.go
new file mode 100644
index 00000000..69e0c56b
--- /dev/null
+++ b/deliberi/deliberi.go
@@ -0,0 +1,87 @@
+package deliberi
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "time"
+
+ "github.com/go-chi/chi/v5"
+ "tangled.org/core/deliberi/config"
+ deldb "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/mailer"
+ delxrpc "tangled.org/core/deliberi/xrpc"
+ "tangled.org/core/idresolver"
+ "tangled.org/core/log"
+ "tangled.org/core/xrpc/serviceauth"
+)
+
+func Run(ctx context.Context, cfg *config.Config) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ logger := log.FromContext(ctx)
+
+ database, err := deldb.Make(ctx, cfg.DbPath)
+ if err != nil {
+ return fmt.Errorf("initializing db: %w", err)
+ }
+
+ resolver := idresolver.DefaultResolver(cfg.PlcUrl)
+
+ bobbin := newBobbinClient(cfg.BobbinApiUrl)
+ ingester, err := NewIngester(database, bobbin, cfg.JetstreamEndpoint, cfg.Hostname, log.SubLogger(logger, "ingest"))
+ if err != nil {
+ return fmt.Errorf("creating ingester: %w", err)
+ }
+ go func() {
+ if err := ingester.Run(ctx); err != nil {
+ logger.Error("ingester stopped", "err", err)
+ cancel()
+ }
+ }()
+
+ sender := mailer.New(cfg.Resend, log.SubLogger(logger, "email"))
+ dispatcher := NewDispatcher(database, sender, cfg.Resend, cfg.BaseURL, resolver, log.SubLogger(logger, "digest"), cfg.Dev)
+ go dispatcher.Start(ctx)
+
+ serviceAuth := serviceauth.NewServiceAuth(logger, resolver.Directory(), serviceauth.DidWeb(cfg.Hostname).String())
+ x := &delxrpc.Xrpc{
+ DB: database,
+ Config: cfg,
+ Logger: log.SubLogger(logger, "xrpc"),
+ ServiceAuth: serviceAuth,
+ IdResolver: resolver,
+ Sender: sender,
+ }
+
+ srv := &http.Server{Addr: cfg.ListenAddr, Handler: chiMount(x)}
+ go func() {
+ logger.Info("starting http server", "addr", cfg.ListenAddr)
+ if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+ logger.Error("http server failed", "err", err)
+ cancel()
+ }
+ }()
+
+ logger.Info("startup complete")
+ <-ctx.Done()
+ logger.Info("received shutdown signal", "reason", ctx.Err())
+
+ shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer shutdownCancel()
+ if err := srv.Shutdown(shutdownCtx); err != nil {
+ logger.Error("http shutdown", "err", err)
+ }
+ if err := database.Close(); err != nil {
+ logger.Error("db close", "err", err)
+ }
+ logger.Info("shutdown complete")
+ return nil
+}
+
+func chiMount(x *delxrpc.Xrpc) http.Handler {
+ mux := chi.NewRouter()
+ mux.Mount("/xrpc", x.Router())
+ return mux
+}
diff --git a/deliberi/digest.go b/deliberi/digest.go
new file mode 100644
index 00000000..d6de611e
--- /dev/null
+++ b/deliberi/digest.go
@@ -0,0 +1,371 @@
+package deliberi
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "html/template"
+ "log/slog"
+ "strings"
+ gotemplate "text/template"
+ "time"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "tangled.org/core/deliberi/config"
+ deldb "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/mailer"
+ "tangled.org/core/deliberi/models"
+ "tangled.org/core/idresolver"
+)
+
+const digestTextTmpl = `Hi {{.RecipientHandle}},
+
+You have {{.Count}} new notification(s) on Tangled:
+
+{{range .Groups}}- {{wrap 70 " " .Header}}{{if .EntityRef}}
+ {{wrap 70 " " .EntityRef}}{{end}}
+ {{.URL}}
+{{end}}{{if .HasMore}}
+View more notifications: {{.NotificationsURL}}
+{{end}}---
+Manage notifications: {{.SettingsURL}}
+`
+
+const digestHTMLTmpl = `
+
+
+
+
+Tangled notifications
+
+
+
+
+
+
+
+
+
+
+
+ 
+ Hi {{.RecipientHandle}},
+
+ You have {{.Count}} new notification{{if gt .Count 1}}s{{end}}:
+
+
+ {{range .Groups}}
+
+
+
+
+
+
+ 
+ |
+
+ {{.HeaderHTML}}
+ {{if .EntityRef}}{{.EntityRef}} {{end}}
+ |
+
+
+
+ |
+
+ {{end}}
+
+ {{if .HasMore}}
+
+ View more notifications
+
+ {{end}}
+
+ |
+
+
+
+ |
+
+
+
+
+`
+
+const digestMaxGroups = 10
+
+type digestGroup struct {
+ IconURL string
+ Header string
+ HeaderHTML template.HTML
+ EntityRef string
+ URL string
+}
+
+type digestData struct {
+ RecipientHandle string
+ Count int
+ Groups []digestGroup
+ HasMore bool
+ NotificationsURL string
+ SettingsURL string
+ AssetsURL string
+}
+
+func notifHeader(n *models.Notification, actor, repo string) string {
+ switch n.Type {
+ case models.NotificationTypeIssueCreated:
+ return actor + " opened an issue on " + repo
+ case models.NotificationTypeIssueCommented:
+ return actor + " commented on an issue on " + repo
+ case models.NotificationTypeIssueClosed:
+ return actor + " closed an issue on " + repo
+ case models.NotificationTypeIssueReopen:
+ return actor + " reopened an issue on " + repo
+ case models.NotificationTypePullCreated:
+ return actor + " created a PR on " + repo
+ case models.NotificationTypePullCommented:
+ return actor + " commented on a PR on " + repo
+ case models.NotificationTypePullMerged:
+ return actor + " merged a PR on " + repo
+ case models.NotificationTypePullClosed:
+ return actor + " closed a PR on " + repo
+ case models.NotificationTypePullReopen:
+ return actor + " reopened a PR on " + repo
+ case models.NotificationTypeUserMentioned:
+ if n.EntityAt != "" && syntax.ATURI(n.EntityAt).Collection().String() == "sh.tangled.repo.pull" {
+ return actor + " mentioned you on a pull request in " + repo
+ }
+ return actor + " mentioned you on an issue in " + repo
+ case models.NotificationTypeIssueAssigned:
+ return actor + " assigned you to an issue on " + repo
+ case models.NotificationTypeIssueUnassigned:
+ return actor + " unassigned you from an issue on " + repo
+ case models.NotificationTypePullAssigned:
+ return actor + " assigned you to a PR on " + repo
+ case models.NotificationTypePullUnassigned:
+ return actor + " unassigned you from a PR on " + repo
+ default:
+ return actor + " updated " + repo
+ }
+}
+
+func notifEntityRef(n *models.Notification) string {
+ return n.EntityTitle
+}
+
+func wordwrap(width int, indent, text string) string {
+ words := strings.Fields(text)
+ if len(words) == 0 {
+ return text
+ }
+ var b strings.Builder
+ col := 0
+ for i, w := range words {
+ if i == 0 {
+ b.WriteString(w)
+ col = len(w)
+ continue
+ }
+ if col+1+len(w) > width {
+ b.WriteString("\n" + indent)
+ b.WriteString(w)
+ col = len(indent) + len(w)
+ } else {
+ b.WriteByte(' ')
+ b.WriteString(w)
+ col += 1 + len(w)
+ }
+ }
+ return b.String()
+}
+
+type Dispatcher struct {
+ db *deldb.DB
+ sender *mailer.Sender
+ baseURL string
+ assetsURL string
+ resolver *idresolver.Resolver
+ logger *slog.Logger
+ batchWait time.Duration
+ interval time.Duration
+
+ textTmpl *gotemplate.Template
+ htmlTmpl *template.Template
+}
+
+func NewDispatcher(database *deldb.DB, sender *mailer.Sender, resend config.ResendConfig, baseURL string, resolver *idresolver.Resolver, logger *slog.Logger, dev bool) *Dispatcher {
+ batchWait := 10 * time.Minute
+ interval := 5 * time.Minute
+ if dev {
+ batchWait = 30 * time.Second
+ interval = 15 * time.Second
+ }
+ return &Dispatcher{
+ db: database,
+ sender: sender,
+ baseURL: strings.TrimRight(baseURL, "/"),
+ assetsURL: strings.TrimRight(resend.AssetsURL, "/") + "/",
+ resolver: resolver,
+ logger: logger,
+ batchWait: batchWait,
+ interval: interval,
+ textTmpl: gotemplate.Must(gotemplate.New("digest-text").Funcs(gotemplate.FuncMap{"wrap": wordwrap}).Parse(digestTextTmpl)),
+ htmlTmpl: template.Must(template.New("digest-html").Parse(digestHTMLTmpl)),
+ }
+}
+
+func (d *Dispatcher) Start(ctx context.Context) {
+ d.logger.Info("email dispatcher started", "interval", d.interval, "batchWait", d.batchWait)
+ ticker := time.NewTicker(d.interval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ d.dispatch(ctx)
+ case <-ctx.Done():
+ d.logger.Info("email dispatcher stopped")
+ return
+ }
+ }
+}
+
+func (d *Dispatcher) dispatch(ctx context.Context) {
+ cutoff := time.Now().Add(-d.batchWait)
+ recipients, err := deldb.GetPendingEmailDigestRecipients(d.db, cutoff)
+ if err != nil {
+ d.logger.Error("email dispatcher: failed to get recipients", "err", err)
+ return
+ }
+ for _, did := range recipients {
+ if err := d.sendDigest(ctx, did, cutoff); err != nil {
+ d.logger.Error("email dispatcher: failed to send digest", "did", did, "err", err)
+ }
+ }
+}
+
+func (d *Dispatcher) sendDigest(ctx context.Context, recipientDid string, cutoff time.Time) error {
+ em, err := deldb.GetPrimaryEmail(d.db, recipientDid)
+ if err != nil || !em.Verified {
+ return nil
+ }
+
+ notifs, err := deldb.GetPendingNotificationsForEmailDigest(d.db, recipientDid, cutoff)
+ if err != nil {
+ return fmt.Errorf("get pending notifications: %w", err)
+ }
+ if len(notifs) == 0 {
+ return nil
+ }
+
+ handle := recipientDid
+ if id, err := d.resolver.ResolveIdent(ctx, recipientDid); err == nil && !id.Handle.IsInvalidHandle() {
+ handle = id.Handle.String()
+ }
+
+ subject, text, html, err := d.renderDigest(ctx, handle, notifs)
+ if err != nil {
+ return fmt.Errorf("render digest: %w", err)
+ }
+
+ // collect ids before sending so notifications arriving mid-send are not
+ // marked emailed.
+ ids := make([]int64, len(notifs))
+ for i, n := range notifs {
+ ids[i] = n.ID
+ }
+
+ if err := d.sender.Send(em.Address, subject, text, html); err != nil {
+ return fmt.Errorf("send email: %w", err)
+ }
+
+ d.logger.Info("email dispatcher: digest sent", "did", recipientDid, "notifications", len(notifs))
+
+ if err := deldb.MarkEmailed(d.db, ids); err != nil {
+ d.logger.Error("email dispatcher: failed to mark emailed", "did", recipientDid, "err", err)
+ }
+ return nil
+}
+
+func (d *Dispatcher) renderDigest(ctx context.Context, recipientHandle string, notifs []*models.Notification) (subject, text, html string, err error) {
+ count := len(notifs)
+
+ shown := notifs
+ if len(shown) > digestMaxGroups {
+ shown = shown[:digestMaxGroups]
+ }
+
+ groups := make([]digestGroup, 0, len(shown))
+ for _, n := range shown {
+ actorHandle := n.ActorDid
+ if id, err2 := d.resolver.ResolveIdent(ctx, n.ActorDid); err2 == nil && !id.Handle.IsInvalidHandle() {
+ actorHandle = id.Handle.String()
+ }
+
+ repoStr := ""
+ if n.RepoDid != "" {
+ repoHandle := n.RepoDid
+ if id, err2 := d.resolver.ResolveIdent(ctx, n.RepoDid); err2 == nil && !id.Handle.IsInvalidHandle() {
+ repoHandle = id.Handle.String()
+ }
+ if n.RepoName != "" {
+ repoStr = repoHandle + "/" + n.RepoName
+ } else {
+ repoStr = repoHandle
+ }
+ }
+
+ header := notifHeader(n, actorHandle, repoStr)
+ headerHTML := template.HTML(strings.Replace(header, actorHandle, ""+actorHandle+"", 1))
+ groups = append(groups, digestGroup{
+ IconURL: d.assetsURL + n.Icon() + ".png",
+ Header: header,
+ HeaderHTML: headerHTML,
+ EntityRef: notifEntityRef(n),
+ URL: d.baseURL + n.URL(d.resolver),
+ })
+ }
+
+ data := digestData{
+ RecipientHandle: recipientHandle,
+ Count: count,
+ Groups: groups,
+ HasMore: count > digestMaxGroups,
+ NotificationsURL: d.baseURL + "/notifications",
+ SettingsURL: d.baseURL + "/settings/notifications",
+ AssetsURL: d.assetsURL,
+ }
+
+ if count > digestMaxGroups {
+ subject = fmt.Sprintf("[%s] %d+ notifications", recipientHandle, digestMaxGroups)
+ } else {
+ subject = fmt.Sprintf("[%s] %d notification(s)", recipientHandle, count)
+ }
+
+ var textBuf bytes.Buffer
+ if err = d.textTmpl.Execute(&textBuf, data); err != nil {
+ return
+ }
+ text = textBuf.String()
+
+ var htmlBuf bytes.Buffer
+ if err = d.htmlTmpl.Execute(&htmlBuf, data); err != nil {
+ return
+ }
+ html = htmlBuf.String()
+ return
+}
diff --git a/deliberi/ingest.go b/deliberi/ingest.go
new file mode 100644
index 00000000..56935662
--- /dev/null
+++ b/deliberi/ingest.go
@@ -0,0 +1,243 @@
+package deliberi
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "log/slog"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ jmodels "github.com/bluesky-social/jetstream/pkg/models"
+ "tangled.org/core/api/tangled"
+ deldb "tangled.org/core/deliberi/db"
+ models "tangled.org/core/deliberi/models"
+ js "tangled.org/core/jetstream"
+)
+
+type Ingester struct {
+ db *deldb.DB
+ recipients recipientResolver
+ jc *js.JetstreamClient
+ logger *slog.Logger
+}
+
+var ingestCollections = []string{
+ tangled.RepoNSID,
+ tangled.RepoIssueNSID,
+ tangled.RepoPullNSID,
+ tangled.FeedCommentNSID,
+ tangled.FeedStarNSID,
+ tangled.GraphFollowNSID,
+}
+
+func NewIngester(database *deldb.DB, recipients recipientResolver, endpoint, ident string, logger *slog.Logger) (*Ingester, error) {
+ jc, err := js.NewJetstreamClient(endpoint, ident, ingestCollections, nil, logger, database, false, false)
+ if err != nil {
+ return nil, fmt.Errorf("creating jetstream client: %w", err)
+ }
+ return &Ingester{
+ db: database,
+ recipients: recipients,
+ jc: jc,
+ logger: logger,
+ }, nil
+}
+
+func (i *Ingester) Run(ctx context.Context) error {
+ return i.jc.StartJetstream(ctx, i.process)
+}
+
+func (i *Ingester) process(ctx context.Context, e *jmodels.Event) error {
+ if e.Kind != jmodels.EventKindCommit || e.Commit == nil {
+ return nil
+ }
+ if e.Commit.Operation != jmodels.CommitOperationCreate && e.Commit.Operation != jmodels.CommitOperationUpdate {
+ return nil
+ }
+
+ actorDid := e.Did
+ entityAt := fmt.Sprintf("at://%s/%s/%s", e.Did, e.Commit.Collection, e.Commit.RKey)
+
+ switch e.Commit.Collection {
+ case tangled.RepoNSID:
+ var rec tangled.Repo
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding repo record", "err", err, "uri", entityAt)
+ return nil
+ }
+ ownerDid := e.Did
+ repoDid := ownerDid
+ name := ""
+ if rec.Name != nil {
+ name = *rec.Name
+ }
+ if rec.RepoDid != nil && *rec.RepoDid != "" {
+ repoDid = *rec.RepoDid
+ }
+ if err := deldb.PutRepoName(i.db, repoDid, ownerDid, name); err != nil {
+ i.logger.Warn("caching repo name", "err", err, "repoDid", repoDid, "ownerDid", ownerDid)
+ }
+
+ case tangled.RepoIssueNSID:
+ var rec tangled.RepoIssue
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding issue record", "err", err, "uri", entityAt)
+ return nil
+ }
+ if err := deldb.PutEntityTitle(i.db, entityAt, rec.Title); err != nil {
+ i.logger.Warn("caching entity title", "err", err, "uri", entityAt)
+ }
+ i.notifyEntity(ctx, actorDid, entityAt, entityAt, rec.Repo, models.NotificationTypeIssueCreated, rec.Title, rec.Mentions)
+
+ case tangled.RepoPullNSID:
+ var rec tangled.RepoPull
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding pull record", "err", err, "uri", entityAt)
+ return nil
+ }
+ repoDid := ""
+ if rec.Target != nil {
+ repoDid = rec.Target.Repo
+ }
+ if err := deldb.PutEntityTitle(i.db, entityAt, rec.Title); err != nil {
+ i.logger.Warn("caching entity title", "err", err, "uri", entityAt)
+ }
+ i.notifyEntity(ctx, actorDid, entityAt, entityAt, repoDid, models.NotificationTypePullCreated, rec.Title, rec.Mentions)
+
+ case tangled.FeedCommentNSID:
+ var rec tangled.FeedComment
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding comment record", "err", err, "uri", entityAt)
+ return nil
+ }
+ if rec.Subject == nil {
+ return nil
+ }
+ subjectUri := rec.Subject.Uri
+ var t models.NotificationType
+ switch syntax.ATURI(subjectUri).Collection().String() {
+ case tangled.RepoIssueNSID:
+ t = models.NotificationTypeIssueCommented
+ case tangled.RepoPullNSID:
+ t = models.NotificationTypePullCommented
+ default:
+ return nil
+ }
+ // comment carries no repo did and no mentions field; leave both empty.
+ title := deldb.GetEntityTitle(i.db, subjectUri)
+ i.notifyEntity(ctx, actorDid, entityAt, subjectUri, "", t, title, nil)
+
+ case tangled.FeedStarNSID:
+ var rec tangled.FeedStar
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding star record", "err", err, "uri", entityAt)
+ return nil
+ }
+ if rec.Subject == nil || rec.Subject.FeedStar_Repo == nil {
+ return nil
+ }
+ repoDid := rec.Subject.FeedStar_Repo.Did
+ recipientDid := i.hydrateRepoOwner(ctx, repoDid)
+ if recipientDid == "" {
+ i.logger.Warn("star: could not resolve repo owner, skipping notification", "repoDid", repoDid)
+ return nil
+ }
+ // stars notify the repo owner directly, no fanout.
+ i.notifyOne(ctx, recipientDid, actorDid, entityAt, "", repoDid, models.NotificationTypeRepoStarred, "")
+
+ case tangled.GraphFollowNSID:
+ var rec tangled.GraphFollow
+ if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
+ i.logger.Warn("decoding follow record", "err", err, "uri", entityAt)
+ return nil
+ }
+ if rec.Subject == "" {
+ return nil
+ }
+ i.notifyOne(ctx, rec.Subject, actorDid, entityAt, "", "", models.NotificationTypeFollowed, "")
+ }
+
+ return nil
+}
+
+func (i *Ingester) notifyEntity(ctx context.Context, actorDid, sourceAt, entityAt, repoDid string, t models.NotificationType, title string, mentions []string) {
+ seen := make(map[string]struct{})
+
+ subscribers, err := i.recipients.ListRecipients(ctx, entityAt)
+ if err != nil {
+ i.logger.Warn("listing recipients", "err", err, "entity", entityAt)
+ }
+
+ for _, dids := range [][]string{subscribers, mentions} {
+ for _, did := range dids {
+ if _, ok := seen[did]; ok {
+ continue
+ }
+ seen[did] = struct{}{}
+ i.deliver(did, actorDid, sourceAt, entityAt, repoDid, t, title)
+ }
+ }
+}
+
+// hydrateRepoOwner reads the repo cache, falling back to bobbin on a miss so
+// stars on repos the ingester never saw still notify their owner. The resolved
+// mapping is cached, so the cache fills in as repos are starred rather than
+// needing a backfill.
+func (i *Ingester) hydrateRepoOwner(ctx context.Context, repoDid string) string {
+ if owner := deldb.GetRepoOwner(i.db, repoDid); owner != "" {
+ return owner
+ }
+ if i.recipients == nil {
+ return ""
+ }
+
+ owner, name, err := i.recipients.RepoOwner(ctx, repoDid)
+ if err != nil {
+ i.logger.Warn("resolving repo owner", "err", err, "repoDid", repoDid)
+ return ""
+ }
+ if owner == "" {
+ return ""
+ }
+
+ // a nameless record must not blank out a name we already cached.
+ if name == "" {
+ name = deldb.GetRepoName(i.db, repoDid)
+ }
+ if err := deldb.PutRepoName(i.db, repoDid, owner, name); err != nil {
+ i.logger.Warn("caching repo owner", "err", err, "repoDid", repoDid)
+ }
+ return owner
+}
+
+func (i *Ingester) notifyOne(ctx context.Context, recipientDid, actorDid, sourceAt, entityAt, repoDid string, t models.NotificationType, title string) {
+ i.deliver(recipientDid, actorDid, sourceAt, entityAt, repoDid, t, title)
+}
+
+func (i *Ingester) deliver(recipientDid, actorDid, sourceAt, entityAt, repoDid string, t models.NotificationType, title string) {
+ if recipientDid == "" || recipientDid == actorDid {
+ return
+ }
+
+ prefs, err := deldb.GetNotificationPreference(i.db, recipientDid)
+ if err != nil {
+ i.logger.Warn("loading prefs", "err", err, "recipient", recipientDid)
+ return
+ }
+ if !prefs.ShouldNotify(t) {
+ return
+ }
+
+ n := &models.Notification{
+ RecipientDid: recipientDid,
+ AtUri: sourceAt,
+ Type: t,
+ ActorDid: actorDid,
+ RepoDid: repoDid,
+ EntityAt: entityAt,
+ EntityTitle: title,
+ }
+ if err := deldb.CreateNotification(i.db, n); err != nil {
+ i.logger.Warn("creating notification", "err", err, "recipient", recipientDid, "uri", sourceAt)
+ }
+}
diff --git a/deliberi/ingest_test.go b/deliberi/ingest_test.go
new file mode 100644
index 00000000..eb3bfbae
--- /dev/null
+++ b/deliberi/ingest_test.go
@@ -0,0 +1,275 @@
+package deliberi
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "log/slog"
+ "path/filepath"
+ "testing"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ jmodels "github.com/bluesky-social/jetstream/pkg/models"
+ "tangled.org/core/api/tangled"
+ deldb "tangled.org/core/deliberi/db"
+ models "tangled.org/core/deliberi/models"
+)
+
+type fakeResolver struct {
+ dids []string
+ err error
+
+ // repo owner lookups, keyed by repo did.
+ owners map[string]string
+ repoNames map[string]string
+ ownerErr error
+ // ownerCalls counts RepoOwner calls, to prove cache hits skip the network.
+ ownerCalls *int
+}
+
+func (f fakeResolver) ListRecipients(ctx context.Context, uri string) ([]string, error) {
+ return f.dids, f.err
+}
+
+func (f fakeResolver) RepoOwner(ctx context.Context, repoDid string) (string, string, error) {
+ if f.ownerCalls != nil {
+ *f.ownerCalls++
+ }
+ if f.ownerErr != nil {
+ return "", "", f.ownerErr
+ }
+ return f.owners[repoDid], f.repoNames[repoDid], nil
+}
+
+func starEvent(t *testing.T, actorDid, repoDid, rkey string) *jmodels.Event {
+ t.Helper()
+ raw, err := json.Marshal(tangled.FeedStar{
+ CreatedAt: "2026-01-01T00:00:00Z",
+ Subject: &tangled.FeedStar_Subject{
+ FeedStar_Repo: &tangled.FeedStar_Repo{Did: repoDid},
+ },
+ })
+ if err != nil {
+ t.Fatalf("marshal star: %v", err)
+ }
+ return &jmodels.Event{
+ Did: actorDid,
+ Kind: jmodels.EventKindCommit,
+ Commit: &jmodels.Commit{
+ Operation: jmodels.CommitOperationCreate,
+ Collection: tangled.FeedStarNSID,
+ RKey: rkey,
+ Record: raw,
+ },
+ }
+}
+
+func newTestIngester(t *testing.T, r recipientResolver) *Ingester {
+ t.Helper()
+ database, err := deldb.Make(context.Background(), filepath.Join(t.TempDir(), "x.db"))
+ if err != nil {
+ t.Fatalf("make db: %v", err)
+ }
+ t.Cleanup(func() { database.Close() })
+ return &Ingester{
+ db: database,
+ recipients: r,
+ logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
+ }
+}
+
+func countFor(t *testing.T, i *Ingester, did string) int64 {
+ t.Helper()
+ n, err := deldb.CountNotifications(i.db, did)
+ if err != nil {
+ t.Fatalf("count: %v", err)
+ }
+ return n
+}
+
+func TestNotifyEntitySubscriberGetsRow(t *testing.T) {
+ i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}})
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil)
+ if got := countFor(t, i, "did:sub"); got != 1 {
+ t.Fatalf("subscriber rows = %d, want 1", got)
+ }
+}
+
+func TestActorNeverNotified(t *testing.T) {
+ i := newTestIngester(t, fakeResolver{dids: []string{"did:actor"}})
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:actor"})
+ if got := countFor(t, i, "did:actor"); got != 0 {
+ t.Fatalf("actor rows = %d, want 0", got)
+ }
+}
+
+func TestMentionDeliveredOnResolverError(t *testing.T) {
+ i := newTestIngester(t, fakeResolver{err: io.ErrUnexpectedEOF})
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", []string{"did:mention"})
+ if got := countFor(t, i, "did:mention"); got != 1 {
+ t.Fatalf("mention rows = %d, want 1", got)
+ }
+}
+
+func TestCreateNotificationDedupe(t *testing.T) {
+ i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}})
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil)
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil)
+ if got := countFor(t, i, "did:sub"); got != 1 {
+ t.Fatalf("deduped rows = %d, want 1", got)
+ }
+}
+
+func TestStarNotifiesRepoOwner(t *testing.T) {
+ calls := 0
+ i := newTestIngester(t, fakeResolver{ownerCalls: &calls})
+ const repoDid = "did:plc:therepo"
+ const ownerDid = "did:plc:bob"
+ defer func() {
+ if calls != 0 {
+ t.Errorf("RepoOwner calls = %d, want 0 (cache hit must not hit bobbin)", calls)
+ }
+ }()
+
+ // Seed the repo DID → owner DID mapping (what the repo handler now does).
+ if err := deldb.PutRepoName(i.db, repoDid, ownerDid, "my-repo"); err != nil {
+ t.Fatalf("PutRepoName: %v", err)
+ }
+
+ // Construct a firehose event for Alice starring Bob's repo.
+ raw, err := json.Marshal(tangled.FeedStar{
+ CreatedAt: "2026-01-01T00:00:00Z",
+ Subject: &tangled.FeedStar_Subject{
+ FeedStar_Repo: &tangled.FeedStar_Repo{Did: repoDid},
+ },
+ })
+ if err != nil {
+ t.Fatalf("marshal star: %v", err)
+ }
+ ev := &jmodels.Event{
+ Did: "did:plc:alice",
+ Kind: jmodels.EventKindCommit,
+ Commit: &jmodels.Commit{
+ Operation: jmodels.CommitOperationCreate,
+ Collection: tangled.FeedStarNSID,
+ RKey: "star1",
+ Record: raw,
+ },
+ }
+
+ if err := i.process(context.Background(), ev); err != nil {
+ t.Fatalf("process: %v", err)
+ }
+
+ if got := countFor(t, i, ownerDid); got != 1 {
+ t.Fatalf("owner rows = %d, want 1", got)
+ }
+ if got := countFor(t, i, repoDid); got != 0 {
+ t.Fatalf("repo DID rows = %d, want 0 (notification must go to owner, not repo)", got)
+ }
+ if got := countFor(t, i, "did:plc:alice"); got != 0 {
+ t.Fatalf("actor rows = %d, want 0 (star author must not self-notify)", got)
+ }
+}
+
+func TestStarResolvesOwnerFromBobbinOnCacheMiss(t *testing.T) {
+ const repoDid = "did:plc:therepo"
+ const ownerDid = "did:plc:bob"
+
+ // nothing cached: the owner must come from bobbin.
+ i := newTestIngester(t, fakeResolver{
+ owners: map[string]string{repoDid: ownerDid},
+ repoNames: map[string]string{repoDid: "my-repo"},
+ })
+
+ if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil {
+ t.Fatalf("process: %v", err)
+ }
+
+ if got := countFor(t, i, ownerDid); got != 1 {
+ t.Fatalf("owner rows = %d, want 1", got)
+ }
+ // the lookup must be cached, so the next star on this repo is a local hit.
+ if got := deldb.GetRepoOwner(i.db, repoDid); got != ownerDid {
+ t.Errorf("cached owner = %q, want %q", got, ownerDid)
+ }
+ if got := deldb.GetRepoName(i.db, repoDid); got != "my-repo" {
+ t.Errorf("cached name = %q, want %q", got, "my-repo")
+ }
+}
+
+func TestStarKeepsCachedNameWhenRecordIsNameless(t *testing.T) {
+ const repoDid = "did:plc:therepo"
+ const ownerDid = "did:plc:bob"
+
+ // a row with a name but no owner, as migrated rows can have.
+ i := newTestIngester(t, fakeResolver{owners: map[string]string{repoDid: ownerDid}})
+ if err := deldb.PutRepoName(i.db, repoDid, "", "my-repo"); err != nil {
+ t.Fatalf("PutRepoName: %v", err)
+ }
+
+ if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil {
+ t.Fatalf("process: %v", err)
+ }
+
+ if got := deldb.GetRepoName(i.db, repoDid); got != "my-repo" {
+ t.Errorf("cached name = %q, want %q (nameless record must not blank it)", got, "my-repo")
+ }
+}
+
+func TestStarSkipsWhenOwnerUnresolvable(t *testing.T) {
+ // nothing cached and bobbin is unreachable, so the star handler should skip.
+ i := newTestIngester(t, fakeResolver{ownerErr: io.ErrUnexpectedEOF})
+ const repoDid = "did:plc:therepo"
+
+ if err := i.process(context.Background(), starEvent(t, "did:plc:alice", repoDid, "star1")); err != nil {
+ t.Fatalf("process: %v", err)
+ }
+
+ if got := countFor(t, i, "did:plc:bob"); got != 0 {
+ t.Fatalf("bob rows = %d, want 0 (unresolvable owner => skip)", got)
+ }
+}
+
+func TestRepoHandlerSeedsOwner(t *testing.T) {
+ // Verify that the ingester's repo processing stores the correct mapping:
+ // repo_did → owner_did and name.
+ // This test exercises the same code path as the repo NSID case in process().
+ db, err := deldb.Make(context.Background(), filepath.Join(t.TempDir(), "x.db"))
+ if err != nil {
+ t.Fatalf("make db: %v", err)
+ }
+ t.Cleanup(func() { db.Close() })
+
+ const ownerDid = "did:plc:owner"
+ const repoDid = "did:plc:therepo"
+ const repoName = "my-repo"
+
+ if err := deldb.PutRepoName(db, repoDid, ownerDid, repoName); err != nil {
+ t.Fatalf("PutRepoName: %v", err)
+ }
+
+ if got := deldb.GetRepoOwner(db, repoDid); got != ownerDid {
+ t.Fatalf("GetRepoOwner = %q, want %q", got, ownerDid)
+ }
+ if got := deldb.GetRepoName(db, repoDid); got != repoName {
+ t.Fatalf("GetRepoName = %q, want %q", got, repoName)
+ }
+ // Owner DID alone should NOT resolve as a repo name (it's not the key).
+ if got := deldb.GetRepoName(db, ownerDid); got != "" {
+ t.Fatalf("GetRepoName(ownerDid) = %q, want empty (owner is not the cache key)", got)
+ }
+}
+
+func TestDisabledPrefSuppressesRow(t *testing.T) {
+ i := newTestIngester(t, fakeResolver{dids: []string{"did:sub"}})
+ prefs := models.DefaultNotificationPreferences(syntax.DID("did:sub"))
+ prefs.IssueCreated = false
+ if err := deldb.UpsertNotificationPreferences(i.db, prefs); err != nil {
+ t.Fatalf("upsert prefs: %v", err)
+ }
+ i.notifyEntity(context.Background(), "did:actor", "at://src", "at://entity", "did:repo", models.NotificationTypeIssueCreated, "title", nil)
+ if got := countFor(t, i, "did:sub"); got != 0 {
+ t.Fatalf("disabled-pref rows = %d, want 0", got)
+ }
+}
diff --git a/deliberi/mailer/mailer.go b/deliberi/mailer/mailer.go
new file mode 100644
index 00000000..f2f82823
--- /dev/null
+++ b/deliberi/mailer/mailer.go
@@ -0,0 +1,40 @@
+package mailer
+
+import (
+ "fmt"
+ "log/slog"
+
+ appviewemail "tangled.org/core/appview/email"
+ "tangled.org/core/deliberi/config"
+)
+
+type Sender struct {
+ apiKey string
+ from string
+ logger *slog.Logger
+}
+
+func New(resend config.ResendConfig, logger *slog.Logger) *Sender {
+ return &Sender{apiKey: resend.ApiKey, from: resend.SentFrom, logger: logger}
+}
+
+// Send delivers one email. with no Resend key it prints to stdout instead of
+// sending, and reports success so callers proceed normally in dev.
+func (s *Sender) Send(to, subject, text, html string) error {
+ from := "Tangled <" + s.from + ">"
+
+ if s.apiKey == "" {
+ s.logger.Info("resend api key unset; writing email to stdout", "to", to, "subject", subject)
+ fmt.Printf("\n=== deliberi email (stdout, no resend key) ===\nfrom: %s\nto: %s\nsubject: %s\n\n%s\n===============================================\n\n", from, to, subject, text)
+ return nil
+ }
+
+ return appviewemail.SendEmail(appviewemail.Email{
+ APIKey: s.apiKey,
+ From: from,
+ To: to,
+ Subject: subject,
+ Text: text,
+ Html: html,
+ })
+}
diff --git a/deliberi/models/email.go b/deliberi/models/email.go
new file mode 100644
index 00000000..d3189aba
--- /dev/null
+++ b/deliberi/models/email.go
@@ -0,0 +1,14 @@
+package models
+
+import "time"
+
+type Email struct {
+ ID int64
+ Did string
+ Address string
+ Verified bool
+ Primary bool
+ VerificationCode string
+ LastSent *time.Time
+ CreatedAt time.Time
+}
diff --git a/deliberi/models/notifications.go b/deliberi/models/notifications.go
new file mode 100644
index 00000000..cd195ed2
--- /dev/null
+++ b/deliberi/models/notifications.go
@@ -0,0 +1,216 @@
+package models
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "tangled.org/core/idresolver"
+)
+
+type NotificationType string
+
+const (
+ NotificationTypeRepoStarred NotificationType = "repo_starred"
+ NotificationTypeIssueCreated NotificationType = "issue_created"
+ NotificationTypeIssueCommented NotificationType = "issue_commented"
+ NotificationTypePullCreated NotificationType = "pull_created"
+ NotificationTypePullCommented NotificationType = "pull_commented"
+ NotificationTypeFollowed NotificationType = "followed"
+ NotificationTypePullMerged NotificationType = "pull_merged"
+ NotificationTypeIssueClosed NotificationType = "issue_closed"
+ NotificationTypeIssueReopen NotificationType = "issue_reopen"
+ NotificationTypePullClosed NotificationType = "pull_closed"
+ NotificationTypePullReopen NotificationType = "pull_reopen"
+ NotificationTypeUserMentioned NotificationType = "user_mentioned"
+ NotificationTypeIssueAssigned NotificationType = "issue_assigned"
+ NotificationTypeIssueUnassigned NotificationType = "issue_unassigned"
+ NotificationTypePullAssigned NotificationType = "pull_assigned"
+ NotificationTypePullUnassigned NotificationType = "pull_unassigned"
+)
+
+var SocialNotificationTypes = []NotificationType{
+ NotificationTypeRepoStarred,
+ NotificationTypeFollowed,
+}
+
+var WorkNotificationTypes = []NotificationType{
+ NotificationTypeIssueCreated,
+ NotificationTypeIssueCommented,
+ NotificationTypeIssueClosed,
+ NotificationTypeIssueReopen,
+ NotificationTypePullCreated,
+ NotificationTypePullCommented,
+ NotificationTypePullMerged,
+ NotificationTypePullClosed,
+ NotificationTypePullReopen,
+ NotificationTypeUserMentioned,
+ NotificationTypeIssueAssigned,
+ NotificationTypeIssueUnassigned,
+ NotificationTypePullAssigned,
+ NotificationTypePullUnassigned,
+}
+
+// email digest types; social types (repo_starred, followed) are excluded.
+var EmailNotificationTypes = []NotificationType{
+ NotificationTypeIssueCreated,
+ NotificationTypeIssueCommented,
+ NotificationTypeIssueClosed,
+ NotificationTypeIssueReopen,
+ NotificationTypePullCreated,
+ NotificationTypePullCommented,
+ NotificationTypePullMerged,
+ NotificationTypePullClosed,
+ NotificationTypePullReopen,
+ NotificationTypeUserMentioned,
+ NotificationTypeIssueAssigned,
+ NotificationTypeIssueUnassigned,
+ NotificationTypePullAssigned,
+ NotificationTypePullUnassigned,
+}
+
+type Notification struct {
+ ID int64
+ RecipientDid string
+ AtUri string // source record (comment/issue/pull/star/follow); dedupe key
+ Type NotificationType
+ ActorDid string
+ RepoDid string
+ RepoName string // not stored; filled from the repo cache when the digest reads
+ EntityAt string // related issue/pull at-uri (for links); source itself for issue/pull
+ EntityTitle string
+ Read bool
+ Emailed bool
+ Created time.Time
+}
+
+func (n *Notification) Icon() string {
+ switch n.Type {
+ case NotificationTypeRepoStarred:
+ return "star"
+ case NotificationTypeIssueCreated, NotificationTypeIssueReopen:
+ return "circle-dot"
+ case NotificationTypeIssueCommented, NotificationTypePullCommented:
+ return "message-square"
+ case NotificationTypeIssueClosed:
+ return "ban"
+ case NotificationTypePullCreated, NotificationTypePullReopen:
+ return "git-pull-request-create"
+ case NotificationTypePullMerged:
+ return "git-merge"
+ case NotificationTypePullClosed:
+ return "git-pull-request-closed"
+ case NotificationTypeFollowed:
+ return "user-plus"
+ case NotificationTypeUserMentioned:
+ return "at-sign"
+ case NotificationTypeIssueAssigned, NotificationTypePullAssigned:
+ return "user-round-arrow-forward"
+ case NotificationTypeIssueUnassigned, NotificationTypePullUnassigned:
+ return "user-round-minus"
+ default:
+ return ""
+ }
+}
+
+func (n *Notification) URL(res *idresolver.Resolver) string {
+ resolve := func(did string) string {
+ if id, err := res.ResolveIdent(context.Background(), did); err == nil && !id.Handle.IsInvalidHandle() {
+ return id.Handle.String()
+ }
+ return did
+ }
+
+ if n.Type == NotificationTypeFollowed {
+ return "/" + resolve(n.ActorDid)
+ }
+ if n.RepoDid == "" || n.RepoName == "" {
+ return ""
+ }
+ repoHandle := resolve(n.RepoDid)
+ if n.EntityAt != "" {
+ switch syntax.ATURI(n.EntityAt).Collection().String() {
+ case "sh.tangled.repo.issue":
+ return fmt.Sprintf("/%s/%s/issues/%s", repoHandle, n.RepoName, n.EntityAt)
+ case "sh.tangled.repo.pull":
+ return fmt.Sprintf("/%s/%s/pulls/%s", repoHandle, n.RepoName, n.EntityAt)
+ }
+ }
+ return fmt.Sprintf("/%s/%s", repoHandle, n.RepoName)
+}
+
+func Category(t NotificationType) string {
+ for _, st := range SocialNotificationTypes {
+ if st == t {
+ return "social"
+ }
+ }
+ return "work"
+}
+
+type NotificationPreferences struct {
+ ID int64
+ UserDid syntax.DID
+ RepoStarred bool
+ IssueCreated bool
+ IssueCommented bool
+ PullCreated bool
+ PullCommented bool
+ Followed bool
+ UserMentioned bool
+ PullMerged bool
+ IssueClosed bool
+ EmailNotifications bool
+}
+
+func (prefs *NotificationPreferences) ShouldNotify(t NotificationType) bool {
+ switch t {
+ case NotificationTypeRepoStarred:
+ return prefs.RepoStarred
+ case NotificationTypeIssueCreated:
+ return prefs.IssueCreated
+ case NotificationTypeIssueCommented:
+ return prefs.IssueCommented
+ case NotificationTypeIssueClosed:
+ return prefs.IssueClosed
+ case NotificationTypeIssueReopen:
+ return prefs.IssueCreated
+ case NotificationTypePullCreated:
+ return prefs.PullCreated
+ case NotificationTypePullCommented:
+ return prefs.PullCommented
+ case NotificationTypePullMerged:
+ return prefs.PullMerged
+ case NotificationTypePullClosed:
+ return prefs.PullMerged
+ case NotificationTypePullReopen:
+ return prefs.PullCreated
+ case NotificationTypeFollowed:
+ return prefs.Followed
+ case NotificationTypeUserMentioned,
+ NotificationTypeIssueAssigned,
+ NotificationTypeIssueUnassigned,
+ NotificationTypePullAssigned,
+ NotificationTypePullUnassigned:
+ return prefs.UserMentioned
+ default:
+ return false
+ }
+}
+
+func DefaultNotificationPreferences(user syntax.DID) *NotificationPreferences {
+ return &NotificationPreferences{
+ UserDid: user,
+ RepoStarred: true,
+ IssueCreated: true,
+ IssueCommented: true,
+ PullCreated: true,
+ PullCommented: true,
+ Followed: true,
+ UserMentioned: true,
+ PullMerged: true,
+ IssueClosed: true,
+ EmailNotifications: false,
+ }
+}
diff --git a/deliberi/models/signup.go b/deliberi/models/signup.go
new file mode 100644
index 00000000..6d8be0f5
--- /dev/null
+++ b/deliberi/models/signup.go
@@ -0,0 +1,10 @@
+package models
+
+import "time"
+
+type InflightSignup struct {
+ Id int64
+ Email string
+ InviteCode string
+ Created time.Time
+}
diff --git a/deliberi/xrpc/account.go b/deliberi/xrpc/account.go
new file mode 100644
index 00000000..172a054f
--- /dev/null
+++ b/deliberi/xrpc/account.go
@@ -0,0 +1,122 @@
+package xrpc
+
+import (
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "strings"
+
+ "tangled.org/core/api/tangled"
+ db "tangled.org/core/deliberi/db"
+ xrpcerr "tangled.org/core/xrpc/errors"
+)
+
+func (x *Xrpc) AccountListEmails(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "AccountListEmails")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ emails, err := db.GetAllEmails(x.DB, did)
+ if err != nil {
+ l.Error("failed to get emails", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ items := make([]*tangled.TempAccountListEmails_Email, 0, len(emails))
+ for _, e := range emails {
+ items = append(items, &tangled.TempAccountListEmails_Email{
+ Address: e.Address,
+ Verified: e.Verified,
+ Primary: e.Primary,
+ CreatedAt: e.CreatedAt.UTC().Format(timeFormat),
+ })
+ }
+
+ x.writeJSON(w, &tangled.TempAccountListEmails_Output{Emails: items})
+}
+
+func (x *Xrpc) AccountDeleteEmail(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "AccountDeleteEmail")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ var input tangled.TempAccountDeleteEmail_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+ addr := strings.TrimSpace(input.Email)
+
+ existing, err := db.GetEmail(x.DB, did, addr)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ writeError(w, xrpcErrorTag("EmailNotFound", "the email address is not associated with this account"), http.StatusNotFound)
+ return
+ }
+ l.Error("failed to get email", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ if existing.Primary {
+ writeError(w, xrpcErrorTag("CannotDeletePrimary", "the primary email address cannot be deleted; set another address as primary first"), http.StatusBadRequest)
+ return
+ }
+
+ if err := db.DeleteEmail(x.DB, did, addr); err != nil {
+ l.Error("failed to delete email", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ w.WriteHeader(http.StatusOK)
+}
+
+func (x *Xrpc) AccountSetPrimaryEmail(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "AccountSetPrimaryEmail")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ var input tangled.TempAccountSetPrimaryEmail_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+ addr := strings.TrimSpace(input.Email)
+
+ existing, err := db.GetEmail(x.DB, did, addr)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ writeError(w, xrpcErrorTag("EmailNotFound", "the email address is not associated with this account"), http.StatusNotFound)
+ return
+ }
+ l.Error("failed to get email", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ if !existing.Verified {
+ writeError(w, xrpcErrorTag("EmailNotVerified", "the email address must be verified before it can be made primary"), http.StatusBadRequest)
+ return
+ }
+
+ if err := db.MakeEmailPrimary(x.DB, did, addr); err != nil {
+ l.Error("failed to set primary email", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ w.WriteHeader(http.StatusOK)
+}
diff --git a/deliberi/xrpc/notifications.go b/deliberi/xrpc/notifications.go
new file mode 100644
index 00000000..96daf775
--- /dev/null
+++ b/deliberi/xrpc/notifications.go
@@ -0,0 +1,227 @@
+package xrpc
+
+import (
+ "encoding/json"
+ "net/http"
+ "strconv"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "tangled.org/core/api/tangled"
+ db "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/models"
+ "tangled.org/core/orm"
+ xrpcerr "tangled.org/core/xrpc/errors"
+)
+
+func (x *Xrpc) NotificationList(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationList")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ q := r.URL.Query()
+ filters := []orm.Filter{}
+ if q.Get("read") == "unread" {
+ filters = append(filters, orm.FilterEq("read", 0))
+ }
+ switch q.Get("category") {
+ case "social":
+ filters = append(filters, orm.FilterIn("type", models.SocialNotificationTypes))
+ case "work":
+ filters = append(filters, orm.FilterIn("type", models.WorkNotificationTypes))
+ }
+
+ limit := 50
+ if s := q.Get("limit"); s != "" {
+ if n, err := strconv.Atoi(s); err == nil && n > 0 && n <= 100 {
+ limit = n
+ }
+ }
+
+ notifs, err := db.GetNotifications(x.DB, did, limit, filters...)
+ if err != nil {
+ l.Error("failed to list notifications", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ unreadBase := []orm.Filter{orm.FilterEq("read", 0)}
+ workUnread, _ := db.CountNotifications(x.DB, did, append(unreadBase, orm.FilterIn("type", models.WorkNotificationTypes))...)
+ socialUnread, _ := db.CountNotifications(x.DB, did, append(unreadBase, orm.FilterIn("type", models.SocialNotificationTypes))...)
+
+ items := make([]*tangled.TempNotificationListNotifications_Notification, 0, len(notifs))
+ for _, n := range notifs {
+ item := &tangled.TempNotificationListNotifications_Notification{
+ Uri: n.AtUri,
+ Type: string(n.Type),
+ Category: models.Category(n.Type),
+ ActorDid: n.ActorDid,
+ Read: n.Read,
+ CreatedAt: n.Created.Format(timeFormat),
+ }
+ if n.RepoDid != "" {
+ item.RepoDid = &n.RepoDid
+ }
+ if n.EntityAt != "" {
+ switch syntax.ATURI(n.EntityAt).Collection().String() {
+ case "sh.tangled.repo.issue":
+ item.IssueAt = &n.EntityAt
+ case "sh.tangled.repo.pull":
+ item.PullAt = &n.EntityAt
+ }
+ }
+ items = append(items, item)
+ }
+
+ x.writeJSON(w, &tangled.TempNotificationListNotifications_Output{
+ Notifications: items,
+ SocialUnreadCount: socialUnread,
+ WorkUnreadCount: workUnread,
+ })
+}
+
+func (x *Xrpc) NotificationGetUnreadCount(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationGetUnreadCount")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ count, err := db.CountNotifications(x.DB, did, orm.FilterEq("read", 0))
+ if err != nil {
+ l.Error("failed to count unread notifications", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ x.writeJSON(w, &tangled.TempNotificationGetUnreadCount_Output{Count: count})
+}
+
+func (x *Xrpc) NotificationUpdateSeen(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationUpdateSeen")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ var input tangled.TempNotificationUpdateSeen_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+ if input.Uri == "" {
+ writeError(w, badRequestError("uri is required"), http.StatusBadRequest)
+ return
+ }
+
+ if err := db.MarkRead(x.DB, did, input.Uri, input.Read); err != nil {
+ l.Error("failed to update notification read state", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+}
+
+func (x *Xrpc) NotificationMarkAllRead(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationMarkAllRead")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ if err := db.MarkAllRead(x.DB, did); err != nil {
+ l.Error("failed to mark all read", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+}
+
+func (x *Xrpc) NotificationGetPreferences(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationGetPreferences")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ prefs, err := db.GetNotificationPreference(x.DB, did)
+ if err != nil {
+ l.Error("failed to get notification preferences", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ x.writeJSON(w, &tangled.TempNotificationGetPreferences_Preferences{
+ EmailNotifications: prefs.EmailNotifications,
+ Followed: prefs.Followed,
+ IssueClosed: prefs.IssueClosed,
+ IssueCommented: prefs.IssueCommented,
+ IssueCreated: prefs.IssueCreated,
+ PullCommented: prefs.PullCommented,
+ PullCreated: prefs.PullCreated,
+ PullMerged: prefs.PullMerged,
+ RepoStarred: prefs.RepoStarred,
+ UserMentioned: prefs.UserMentioned,
+ })
+}
+
+func (x *Xrpc) NotificationUpdatePreferences(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "NotificationUpdatePreferences")
+
+ did, ok := actorDid(r)
+ if !ok {
+ writeError(w, xrpcerr.MissingActorDidError, http.StatusForbidden)
+ return
+ }
+
+ var input tangled.TempNotificationUpdatePreferences_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+
+ existing, err := db.GetNotificationPreference(x.DB, did)
+ if err != nil {
+ l.Error("failed to get existing notification preferences", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ prefs := &models.NotificationPreferences{
+ UserDid: syntax.DID(did),
+ RepoStarred: applyBoolPtr(existing.RepoStarred, input.RepoStarred),
+ IssueCreated: applyBoolPtr(existing.IssueCreated, input.IssueCreated),
+ IssueCommented: applyBoolPtr(existing.IssueCommented, input.IssueCommented),
+ IssueClosed: applyBoolPtr(existing.IssueClosed, input.IssueClosed),
+ PullCreated: applyBoolPtr(existing.PullCreated, input.PullCreated),
+ PullCommented: applyBoolPtr(existing.PullCommented, input.PullCommented),
+ PullMerged: applyBoolPtr(existing.PullMerged, input.PullMerged),
+ Followed: applyBoolPtr(existing.Followed, input.Followed),
+ UserMentioned: applyBoolPtr(existing.UserMentioned, input.UserMentioned),
+ EmailNotifications: applyBoolPtr(existing.EmailNotifications, input.EmailNotifications),
+ }
+
+ if err := db.UpsertNotificationPreferences(x.DB, prefs); err != nil {
+ l.Error("failed to update notification preferences", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+}
+
+func applyBoolPtr(existing bool, update *bool) bool {
+ if update != nil {
+ return *update
+ }
+ return existing
+}
diff --git a/deliberi/xrpc/signup.go b/deliberi/xrpc/signup.go
new file mode 100644
index 00000000..98ac41c9
--- /dev/null
+++ b/deliberi/xrpc/signup.go
@@ -0,0 +1,251 @@
+package xrpc
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "regexp"
+
+ "tangled.org/core/api/tangled"
+ appviewemail "tangled.org/core/appview/email"
+ db "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/models"
+)
+
+// subdomainRegex validates the requested pds handle label
+var subdomainRegex = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]{2,61}[a-z0-9])?$`)
+
+func isValidSubdomain(name string) bool {
+ return len(name) >= 4 && len(name) <= 63 && subdomainRegex.MatchString(name)
+}
+
+func (x *Xrpc) AccountBeginSignup(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "AccountBeginSignup")
+
+ // signup needs a pds admin secret (turnstile is enforced upstream)
+ if !x.Config.SignupEnabled() {
+ writeError(w, xrpcErrorTag("SignupDisabled", "signup is not currently enabled"), http.StatusFailedDependency)
+ return
+ }
+
+ var input tangled.TempAccountBeginSignup_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+
+ if !appviewemail.IsValidEmail(input.Email) {
+ writeError(w, xrpcErrorTag("InvalidEmail", "invalid email address"), http.StatusBadRequest)
+ return
+ }
+
+ exists, err := db.CheckEmailExistsAtAll(x.DB, input.Email)
+ if err != nil {
+ l.Error("failed to check email existence", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+ if exists {
+ writeError(w, xrpcErrorTag("EmailAlreadyRegistered", "an account already exists for this email"), http.StatusConflict)
+ return
+ }
+
+ // the verification code is an invite code minted by the PDS
+ code, err := x.pdsCreateInviteCode()
+ if err != nil {
+ l.Error("failed to create invite code", "err", err)
+ writeError(w, errUpstream, http.StatusBadGateway)
+ return
+ }
+
+ if err := db.AddInflightSignup(x.DB, models.InflightSignup{Email: input.Email, InviteCode: code}); err != nil {
+ l.Error("failed to add inflight signup", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ // deliberi owns email now: send the verification code inline (stdout in dev
+ // when no resend key is set).
+ text := "Copy and paste this code below to verify your account on Tangled.\n" + code
+ html := "Copy and paste this code below to verify your account on Tangled.
\n" + code + "
"
+ if err := x.Sender.Send(input.Email, "Verify your Tangled account", text, html); err != nil {
+ l.Error("failed to send verification email", "err", err)
+ writeError(w, errInternal, http.StatusInternalServerError)
+ return
+ }
+
+ w.WriteHeader(http.StatusOK)
+}
+
+func (x *Xrpc) AccountCompleteSignup(w http.ResponseWriter, r *http.Request) {
+ l := x.Logger.With("handler", "AccountCompleteSignup")
+
+ if !x.Config.SignupEnabled() {
+ writeError(w, xrpcErrorTag("SignupDisabled", "signup is not currently enabled"), http.StatusFailedDependency)
+ return
+ }
+
+ var input tangled.TempAccountCompleteSignup_Input
+ if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+ writeError(w, errBadRequestBody, http.StatusBadRequest)
+ return
+ }
+
+ if !isValidSubdomain(input.Username) {
+ writeError(w, xrpcErrorTag("InvalidUsername", "invalid username"), http.StatusBadRequest)
+ return
+ }
+
+ emailAddr, err := db.GetEmailForCode(x.DB, input.Code)
+ if err != nil {
+ l.Error("failed to get email for code", "err", err)
+ writeError(w, xrpcErrorTag("InvalidCode", "invalid or expired verification code"), http.StatusBadRequest)
+ return
+ }
+
+ did, handle, err := x.provisionAccount(input.Username, input.Password, emailAddr, input.Code)
+ if err != nil {
+ l.Error("failed to provision account", "err", err)
+ writeError(w, errUpstream, http.StatusBadGateway)
+ return
+ }
+
+ go func() {
+ if err := db.DeleteInflightSignup(x.DB, emailAddr); err != nil {
+ l.Error("failed to delete inflight signup", "err", err)
+ }
+ }()
+
+ x.writeJSON(w, &tangled.TempAccountCompleteSignup_Output{Did: did, Handle: handle})
+}
+
+// provisionAccount creates the pds account and records its verified primary
+// email, rolling back on failure.
+func (x *Xrpc) provisionAccount(username, password, emailAddr, code string) (did, handle string, err error) {
+ success := false
+ emailAdded := false
+ defer func() {
+ if success {
+ return
+ }
+ x.Logger.Info("rolling back signup", "username", username, "did", did)
+ if did != "" {
+ if derr := x.pdsDeleteAccount(did); derr != nil {
+ x.Logger.Error("failed to roll back PDS account", "err", derr, "did", did)
+ }
+ }
+ if emailAdded {
+ if derr := db.DeleteEmail(x.DB, did, emailAddr); derr != nil {
+ x.Logger.Error("failed to roll back email row", "err", derr, "email", emailAddr)
+ }
+ }
+ }()
+
+ did, handle, err = x.pdsCreateAccount(username, password, emailAddr, code)
+ if err != nil {
+ return "", "", err
+ }
+
+ if err = db.AddEmail(x.DB, models.Email{Did: did, Address: emailAddr, Verified: true, Primary: true}); err != nil {
+ return "", "", err
+ }
+ emailAdded = true
+
+ // sites subdomain auto-claim now belongs elsewhere; deliberi does not own the sites table
+
+ success = true
+ return did, handle, nil
+}
+
+// pdsRequest posts to a pds xrpc endpoint; useAuth sends the admin secret via
+// basic auth. these are unauth'd or admin-authed, so they use raw http.
+func (x *Xrpc) pdsRequest(endpoint string, body any, useAuth bool) (*http.Response, error) {
+ jsonData, err := json.Marshal(body)
+ if err != nil {
+ return nil, err
+ }
+ u := fmt.Sprintf("%s/xrpc/%s", x.Config.Pds.Host, endpoint)
+ req, err := http.NewRequest(http.MethodPost, u, bytes.NewBuffer(jsonData))
+ if err != nil {
+ return nil, err
+ }
+ req.Header.Set("Content-Type", "application/json")
+ if useAuth {
+ req.SetBasicAuth("admin", x.Config.Pds.AdminSecret)
+ }
+ return http.DefaultClient.Do(req)
+}
+
+func pdsError(resp *http.Response, action string) error {
+ var e struct {
+ Error string `json:"error"`
+ Message string `json:"message"`
+ }
+ b, _ := io.ReadAll(resp.Body)
+ if err := json.Unmarshal(b, &e); err == nil && e.Message != "" {
+ return fmt.Errorf("failed to %s: %s - %s", action, e.Error, e.Message)
+ }
+ return fmt.Errorf("failed to %s, status %d", action, resp.StatusCode)
+}
+
+func (x *Xrpc) pdsCreateInviteCode() (string, error) {
+ resp, err := x.pdsRequest("com.atproto.server.createInviteCode", map[string]any{"useCount": 1}, true)
+ if err != nil {
+ return "", err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return "", pdsError(resp, "create invite code")
+ }
+ var result map[string]string
+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+ return "", fmt.Errorf("failed to decode invite code response: %w", err)
+ }
+ return result["code"], nil
+}
+
+func (x *Xrpc) pdsCreateAccount(username, password, emailAddr, code string) (did, handle string, err error) {
+ parsed, err := url.Parse(x.Config.Pds.Host)
+ if err != nil {
+ return "", "", fmt.Errorf("invalid PDS host URL: %w", err)
+ }
+ handle = fmt.Sprintf("%s.%s", username, parsed.Hostname())
+
+ body := map[string]string{
+ "email": emailAddr,
+ "handle": handle,
+ "password": password,
+ "inviteCode": code,
+ }
+ resp, err := x.pdsRequest("com.atproto.server.createAccount", body, false)
+ if err != nil {
+ return "", "", err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return "", "", pdsError(resp, "create account")
+ }
+
+ var result struct {
+ DID string `json:"did"`
+ }
+ if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+ return "", "", fmt.Errorf("failed to decode create account response: %w", err)
+ }
+ return result.DID, handle, nil
+}
+
+func (x *Xrpc) pdsDeleteAccount(did string) error {
+ resp, err := x.pdsRequest("com.atproto.admin.deleteAccount", map[string]string{"did": did}, true)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return pdsError(resp, "delete account")
+ }
+ return nil
+}
diff --git a/deliberi/xrpc/xrpc.go b/deliberi/xrpc/xrpc.go
new file mode 100644
index 00000000..29b9f90f
--- /dev/null
+++ b/deliberi/xrpc/xrpc.go
@@ -0,0 +1,132 @@
+package xrpc
+
+import (
+ "encoding/json"
+ "log/slog"
+ "net/http"
+ "runtime/debug"
+
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "github.com/go-chi/chi/v5"
+ "tangled.org/core/api/tangled"
+ config "tangled.org/core/deliberi/config"
+ db "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/mailer"
+ "tangled.org/core/idresolver"
+ xrpcerr "tangled.org/core/xrpc/errors"
+ "tangled.org/core/xrpc/serviceauth"
+)
+
+const ActorDid = serviceauth.ActorDid
+
+type Xrpc struct {
+ DB *db.DB
+ Config *config.Config
+ Logger *slog.Logger
+ ServiceAuth *serviceauth.ServiceAuth
+ IdResolver *idresolver.Resolver
+ Sender *mailer.Sender
+}
+
+func (x *Xrpc) Router() http.Handler {
+ r := chi.NewRouter()
+ r.Use(x.cors)
+
+ // health check, atproto _health convention
+ r.Get("/_health", x.health)
+
+ // open endpoints: signup happens pre-identity, so no service auth
+ r.Post("/"+tangled.TempAccountBeginSignupNSID, x.AccountBeginSignup)
+ r.Post("/"+tangled.TempAccountCompleteSignupNSID, x.AccountCompleteSignup)
+
+ // authenticated endpoints
+ r.Group(func(r chi.Router) {
+ r.Use(x.ServiceAuth.VerifyServiceAuth)
+
+ // notifications
+ r.Get("/"+tangled.TempNotificationListNotificationsNSID, x.NotificationList)
+ r.Get("/"+tangled.TempNotificationGetUnreadCountNSID, x.NotificationGetUnreadCount)
+ r.Post("/"+tangled.TempNotificationUpdateSeenNSID, x.NotificationUpdateSeen)
+ r.Post("/"+tangled.TempNotificationMarkAllReadNSID, x.NotificationMarkAllRead)
+ r.Get("/"+tangled.TempNotificationGetPreferencesNSID, x.NotificationGetPreferences)
+ r.Post("/"+tangled.TempNotificationUpdatePreferencesNSID, x.NotificationUpdatePreferences)
+
+ // account management
+ r.Get("/"+tangled.TempAccountListEmailsNSID, x.AccountListEmails)
+ r.Post("/"+tangled.TempAccountDeleteEmailNSID, x.AccountDeleteEmail)
+ r.Post("/"+tangled.TempAccountSetPrimaryEmailNSID, x.AccountSetPrimaryEmail)
+ })
+
+ return r
+}
+
+// timeFormat is the datetime format used across lexicon output fields
+const timeFormat = "2006-01-02T15:04:05.000Z"
+
+// health responds to /xrpc/_health with the running version
+func (x *Xrpc) health(w http.ResponseWriter, r *http.Request) {
+ x.writeJSON(w, map[string]string{"version": serviceVersion()})
+}
+
+func serviceVersion() string {
+ info, ok := debug.ReadBuildInfo()
+ if !ok {
+ return "dev"
+ }
+ for _, s := range info.Settings {
+ if s.Key == "vcs.revision" && s.Value != "" {
+ return s.Value
+ }
+ }
+ return "dev"
+}
+
+// cors allows the browser origin to call the xrpc endpoints. auth is via
+// bearer tokens, not cookies, so a wildcard origin is safe.
+func (x *Xrpc) cors(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Access-Control-Allow-Origin", "*")
+ w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+ w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")
+ w.Header().Set("Access-Control-Max-Age", "86400")
+ if r.Method == http.MethodOptions {
+ w.WriteHeader(http.StatusNoContent)
+ return
+ }
+ next.ServeHTTP(w, r)
+ })
+}
+
+func writeError(w http.ResponseWriter, e xrpcerr.XrpcError, status int) {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(status)
+ json.NewEncoder(w).Encode(e)
+}
+
+func (x *Xrpc) writeJSON(w http.ResponseWriter, v any) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(v)
+}
+
+func actorDid(r *http.Request) (string, bool) {
+ did, ok := r.Context().Value(ActorDid).(syntax.DID)
+ if !ok {
+ return "", false
+ }
+ return did.String(), true
+}
+
+// stable client-facing errors; handlers log the real cause and return these
+var (
+ errInternal = xrpcErrorTag("InternalError", "internal server error")
+ errBadRequestBody = xrpcErrorTag("InvalidRequest", "invalid request body")
+ errUpstream = xrpcErrorTag("UpstreamError", "an upstream service failed")
+)
+
+func xrpcErrorTag(tag, message string) xrpcerr.XrpcError {
+ return xrpcerr.NewXrpcError(xrpcerr.WithTag(tag), xrpcerr.WithMessage(message))
+}
+
+func badRequestError(message string) xrpcerr.XrpcError {
+ return xrpcErrorTag("InvalidRequest", message)
+}
diff --git a/deliberi/xrpc/xrpc_test.go b/deliberi/xrpc/xrpc_test.go
new file mode 100644
index 00000000..633b9ae8
--- /dev/null
+++ b/deliberi/xrpc/xrpc_test.go
@@ -0,0 +1,264 @@
+package xrpc
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/bluesky-social/indigo/atproto/atcrypto"
+ "github.com/bluesky-social/indigo/atproto/auth"
+ "github.com/bluesky-social/indigo/atproto/identity"
+ "github.com/bluesky-social/indigo/atproto/syntax"
+ "tangled.org/core/api/tangled"
+ config "tangled.org/core/deliberi/config"
+ db "tangled.org/core/deliberi/db"
+ "tangled.org/core/deliberi/models"
+ "tangled.org/core/orm"
+ "tangled.org/core/xrpc/serviceauth"
+)
+
+const (
+ testActor = "did:plc:tester"
+ testAudience = "did:web:test.example"
+)
+
+// newTestXrpc builds an Xrpc backed by a fresh temp DB, service auth wired to a
+// mock directory holding testActor's key. returns router, db, and a token
+// signer for a given lexicon method.
+func newTestXrpc(t *testing.T) (http.Handler, *db.DB, func(nsid string) string) {
+ t.Helper()
+
+ d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "test.db"))
+ if err != nil {
+ t.Fatalf("db.Make: %v", err)
+ }
+ t.Cleanup(func() { d.Close() })
+
+ priv, err := atcrypto.GeneratePrivateKeyP256()
+ if err != nil {
+ t.Fatalf("generate key: %v", err)
+ }
+ pub, err := priv.PublicKey()
+ if err != nil {
+ t.Fatalf("derive pubkey: %v", err)
+ }
+
+ dir := identity.NewMockDirectory()
+ dir.Insert(identity.Identity{
+ DID: syntax.DID(testActor),
+ Keys: map[string]identity.VerificationMethod{
+ "atproto": {Type: "Multikey", PublicKeyMultibase: pub.Multibase()},
+ },
+ })
+
+ logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+ x := &Xrpc{
+ DB: d,
+ Config: &config.Config{},
+ Logger: logger,
+ ServiceAuth: serviceauth.NewServiceAuth(logger, dir, testAudience),
+ }
+
+ sign := func(nsid string) string {
+ lxm := syntax.NSID(nsid)
+ token, err := auth.SignServiceAuth(syntax.DID(testActor), testAudience, time.Minute, &lxm, priv)
+ if err != nil {
+ t.Fatalf("sign service auth: %v", err)
+ }
+ return token
+ }
+
+ return x.Router(), d, sign
+}
+
+func TestHealth(t *testing.T) {
+ router, _, _ := newTestXrpc(t)
+
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/_health", nil))
+
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200", rec.Code)
+ }
+ var body map[string]string
+ if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+ t.Fatalf("decode: %v; body=%s", err, rec.Body.String())
+ }
+ if body["version"] == "" {
+ t.Fatalf("missing version in %s", rec.Body.String())
+ }
+}
+
+func TestServiceAuthRequired(t *testing.T) {
+ router, _, _ := newTestXrpc(t)
+
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/org.tangled.temp.notification.getUnreadCount", nil))
+
+ if rec.Code != http.StatusForbidden {
+ t.Fatalf("status = %d, want 403 without a service-auth token", rec.Code)
+ }
+}
+
+func TestWrongLexiconTokenRejected(t *testing.T) {
+ router, _, sign := newTestXrpc(t)
+
+ // a token minted for a different method must not authorize this call
+ req := httptest.NewRequest(http.MethodGet, "/org.tangled.temp.notification.getUnreadCount", nil)
+ req.Header.Set("Authorization", "Bearer "+sign("org.tangled.temp.notification.listNotifications"))
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, req)
+
+ if rec.Code != http.StatusForbidden {
+ t.Fatalf("status = %d, want 403 for a token bound to a different method", rec.Code)
+ }
+}
+
+func TestNotificationGetUnreadCount(t *testing.T) {
+ router, d, sign := newTestXrpc(t)
+
+ nsid := "org.tangled.temp.notification.getUnreadCount"
+ call := func() int {
+ req := httptest.NewRequest(http.MethodGet, "/"+nsid, nil)
+ req.Header.Set("Authorization", "Bearer "+sign(nsid))
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String())
+ }
+ var out struct {
+ Count int `json:"count"`
+ }
+ if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
+ t.Fatalf("decode: %v; body=%s", err, rec.Body.String())
+ }
+ return out.Count
+ }
+
+ if got := call(); got != 0 {
+ t.Fatalf("empty db count = %d, want 0", got)
+ }
+
+ uri := "at://did:plc:repo/sh.tangled.repo.issue/abc"
+ if err := db.CreateNotification(d, &models.Notification{
+ RecipientDid: testActor,
+ AtUri: uri,
+ Type: models.NotificationTypeIssueCreated,
+ ActorDid: "did:plc:someone",
+ }); err != nil {
+ t.Fatalf("CreateNotification: %v", err)
+ }
+
+ if got := call(); got != 1 {
+ t.Fatalf("count after one unread = %d, want 1", got)
+ }
+
+ if err := db.MarkRead(d, testActor, uri, true); err != nil {
+ t.Fatalf("MarkRead: %v", err)
+ }
+ if got := call(); got != 0 {
+ t.Fatalf("count after marking read = %d, want 0", got)
+ }
+}
+
+func TestNotificationList(t *testing.T) {
+ router, d, sign := newTestXrpc(t)
+
+ uri := "at://did:plc:repo/sh.tangled.repo.issue/abc"
+ if err := db.CreateNotification(d, &models.Notification{
+ RecipientDid: testActor,
+ AtUri: uri,
+ Type: models.NotificationTypeIssueCreated,
+ ActorDid: "did:plc:someone",
+ RepoDid: "did:plc:repo",
+ EntityAt: uri,
+ EntityTitle: "a bug report",
+ }); err != nil {
+ t.Fatalf("CreateNotification: %v", err)
+ }
+
+ list := func(query string) tangled.TempNotificationListNotifications_Output {
+ nsid := "org.tangled.temp.notification.listNotifications"
+ req := httptest.NewRequest(http.MethodGet, "/"+nsid+query, nil)
+ req.Header.Set("Authorization", "Bearer "+sign(nsid))
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String())
+ }
+ var out tangled.TempNotificationListNotifications_Output
+ if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
+ t.Fatalf("decode: %v; body=%s", err, rec.Body.String())
+ }
+ return out
+ }
+
+ out := list("")
+ if len(out.Notifications) != 1 {
+ t.Fatalf("got %d notifications, want 1", len(out.Notifications))
+ }
+ n := out.Notifications[0]
+ if n.Uri != uri {
+ t.Fatalf("uri = %q, want %q", n.Uri, uri)
+ }
+ if n.Type != string(models.NotificationTypeIssueCreated) {
+ t.Fatalf("type = %q, want issue_created", n.Type)
+ }
+ if n.Category != "work" {
+ t.Fatalf("category = %q, want work", n.Category)
+ }
+ if n.RepoDid == nil || *n.RepoDid != "did:plc:repo" {
+ t.Fatalf("repoDid = %v, want did:plc:repo", n.RepoDid)
+ }
+ if n.IssueAt == nil || *n.IssueAt != uri {
+ t.Fatalf("issueAt = %v, want %q", n.IssueAt, uri)
+ }
+ if out.WorkUnreadCount != 1 {
+ t.Fatalf("workUnreadCount = %d, want 1", out.WorkUnreadCount)
+ }
+
+ if err := db.MarkRead(d, testActor, uri, true); err != nil {
+ t.Fatalf("MarkRead: %v", err)
+ }
+ if out := list("?read=unread"); len(out.Notifications) != 0 {
+ t.Fatalf("unread list after read = %d, want 0", len(out.Notifications))
+ }
+}
+
+func TestUpdateSeenPersists(t *testing.T) {
+ router, d, sign := newTestXrpc(t)
+
+ uri := "at://did:plc:repo/sh.tangled.repo.issue/xyz"
+ if err := db.CreateNotification(d, &models.Notification{
+ RecipientDid: testActor,
+ AtUri: uri,
+ Type: models.NotificationTypeIssueCreated,
+ ActorDid: "did:plc:someone",
+ }); err != nil {
+ t.Fatalf("CreateNotification: %v", err)
+ }
+
+ nsid := "org.tangled.temp.notification.updateSeen"
+ req := httptest.NewRequest(http.MethodPost, "/"+nsid, strings.NewReader(`{"uri":"`+uri+`","read":true}`))
+ req.Header.Set("Authorization", "Bearer "+sign(nsid))
+ rec := httptest.NewRecorder()
+ router.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String())
+ }
+
+ unread, err := db.CountNotifications(d, testActor, orm.FilterEq("read", 0))
+ if err != nil {
+ t.Fatalf("CountNotifications: %v", err)
+ }
+ if unread != 0 {
+ t.Fatalf("unread after updateSeen = %d, want 0", unread)
+ }
+}