bretton.dev
A community based topic aggregation platform built on atproto
feat(db): add aggregator database schema
Create comprehensive database schema for aggregator system with
3 tables, 2 triggers, and optimized indexes.
Tables:
- aggregators: Service declarations indexed from Jetstream
- aggregator_authorizations: Community authorizations
- aggregator_posts: Rate limiting tracking (AppView-only)
Key features:
- Optimized indexes for <5ms authorization checks
- Partial indexes WHERE enabled=true for performance
- Foreign keys with CASCADE delete
- Auto-updating stats via triggers
Triggers:
- update_aggregator_communities_count: Tracks communities_using
- update_aggregator_posts_count: Tracks posts_created
Security:
- Audit trail fields (created_by, disabled_by, disabled_at)
- Unique constraint on (aggregator_did, community_did)
- NOT NULL constraints on required fields
Performance:
- idx_aggregator_auth_lookup: Fast (aggregator, community, enabled) checks
- idx_aggregator_posts_rate_limit: Fast rate limit queries
- idx_aggregators_created_at: Sorting by creation date
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
+214
internal/db/migrations/012_create_aggregators_tables.sql
+214
internal/db/migrations/012_create_aggregators_tables.sql
···
1
+
-- +goose Up
2
+
-- Create aggregators tables for indexing aggregator service declarations and authorizations
3
+
-- These records are indexed from Jetstream firehose consumer
4
+
5
+
-- ============================================================================
6
+
-- Table: aggregators
7
+
-- Purpose: Index aggregator service declarations from social.coves.aggregator.service records
8
+
-- Source: Aggregator's own repository (at://aggregator_did/social.coves.aggregator.service/self)
9
+
-- ============================================================================
10
+
CREATE TABLE aggregators (
11
+
-- Primary identity
12
+
did TEXT PRIMARY KEY, -- Aggregator's DID (must match repo DID)
13
+
14
+
-- Service metadata (from lexicon)
15
+
display_name TEXT NOT NULL, -- Human-readable name
16
+
description TEXT, -- What this aggregator does
17
+
config_schema JSONB, -- JSON Schema for community config validation
18
+
avatar_url TEXT, -- Avatar image URL (extracted from blob)
19
+
source_url TEXT, -- URL to source code (transparency)
20
+
maintainer_did TEXT, -- DID of maintainer
21
+
22
+
-- atProto record metadata
23
+
record_uri TEXT NOT NULL UNIQUE, -- AT-URI of service declaration record
24
+
record_cid TEXT NOT NULL, -- CID of current record version
25
+
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- When the aggregator service was created (from lexicon createdAt field)
26
+
indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- When indexed/updated by AppView
27
+
28
+
-- Cached stats (updated by aggregator_posts table triggers/queries)
29
+
communities_using INTEGER NOT NULL DEFAULT 0, -- Count of communities with enabled authorizations
30
+
posts_created BIGINT NOT NULL DEFAULT 0 -- Total posts created by this aggregator
31
+
);
32
+
33
+
-- Indexes for discovery and lookups
34
+
CREATE INDEX idx_aggregators_created_at ON aggregators(created_at DESC);
35
+
CREATE INDEX idx_aggregators_indexed_at ON aggregators(indexed_at DESC);
36
+
CREATE INDEX idx_aggregators_maintainer ON aggregators(maintainer_did);
37
+
38
+
-- Comments
39
+
COMMENT ON TABLE aggregators IS 'Aggregator service declarations indexed from social.coves.aggregator.service records';
40
+
COMMENT ON COLUMN aggregators.did IS 'DID of the aggregator service (matches repo DID)';
41
+
COMMENT ON COLUMN aggregators.config_schema IS 'JSON Schema defining what config options communities can set';
42
+
COMMENT ON COLUMN aggregators.created_at IS 'When the aggregator service was created (from lexicon record createdAt field)';
43
+
COMMENT ON COLUMN aggregators.communities_using IS 'Cached count of communities with enabled=true authorizations';
44
+
45
+
46
+
-- ============================================================================
47
+
-- Table: aggregator_authorizations
48
+
-- Purpose: Index community authorization records for aggregators
49
+
-- Source: Community's repository (at://community_did/social.coves.aggregator.authorization/rkey)
50
+
-- ============================================================================
51
+
CREATE TABLE aggregator_authorizations (
52
+
id BIGSERIAL PRIMARY KEY,
53
+
54
+
-- Authorization identity
55
+
aggregator_did TEXT NOT NULL, -- DID of authorized aggregator
56
+
community_did TEXT NOT NULL, -- DID of community granting access
57
+
58
+
-- Authorization state
59
+
enabled BOOLEAN NOT NULL DEFAULT true, -- Whether aggregator is currently active
60
+
config JSONB, -- Community-specific config (validated against aggregator's schema)
61
+
62
+
-- Audit trail (from lexicon)
63
+
created_at TIMESTAMPTZ NOT NULL, -- When authorization was created
64
+
created_by TEXT NOT NULL, -- DID of moderator who authorized (set by API, not client)
65
+
disabled_at TIMESTAMPTZ, -- When authorization was disabled (if enabled=false)
66
+
disabled_by TEXT, -- DID of moderator who disabled
67
+
68
+
-- atProto record metadata
69
+
record_uri TEXT NOT NULL UNIQUE, -- AT-URI of authorization record
70
+
record_cid TEXT NOT NULL, -- CID of current record version
71
+
indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- When indexed/updated by AppView
72
+
73
+
-- Constraints
74
+
UNIQUE(aggregator_did, community_did), -- One authorization per aggregator per community
75
+
CONSTRAINT fk_aggregator FOREIGN KEY (aggregator_did) REFERENCES aggregators(did) ON DELETE CASCADE,
76
+
CONSTRAINT fk_community FOREIGN KEY (community_did) REFERENCES communities(did) ON DELETE CASCADE
77
+
);
78
+
79
+
-- Indexes for authorization checks (CRITICAL PATH - used on every aggregator post)
80
+
CREATE INDEX idx_aggregator_auth_agg_enabled ON aggregator_authorizations(aggregator_did, enabled) WHERE enabled = true;
81
+
CREATE INDEX idx_aggregator_auth_comm_enabled ON aggregator_authorizations(community_did, enabled) WHERE enabled = true;
82
+
CREATE INDEX idx_aggregator_auth_lookup ON aggregator_authorizations(aggregator_did, community_did, enabled);
83
+
84
+
-- Indexes for listing/discovery
85
+
CREATE INDEX idx_aggregator_auth_agg_did ON aggregator_authorizations(aggregator_did, created_at DESC);
86
+
CREATE INDEX idx_aggregator_auth_comm_did ON aggregator_authorizations(community_did, created_at DESC);
87
+
88
+
-- Comments
89
+
COMMENT ON TABLE aggregator_authorizations IS 'Community authorizations for aggregators indexed from social.coves.aggregator.authorization records';
90
+
COMMENT ON COLUMN aggregator_authorizations.config IS 'Community-specific config, validated against aggregators.config_schema';
91
+
COMMENT ON INDEX idx_aggregator_auth_lookup IS 'CRITICAL: Fast lookup for post creation authorization checks';
92
+
93
+
94
+
-- ============================================================================
95
+
-- Table: aggregator_posts
96
+
-- Purpose: Track posts created by aggregators for rate limiting and stats
97
+
-- Note: This is AppView-only data, not from lexicon records
98
+
-- ============================================================================
99
+
CREATE TABLE aggregator_posts (
100
+
id BIGSERIAL PRIMARY KEY,
101
+
102
+
-- Post identity
103
+
aggregator_did TEXT NOT NULL, -- DID of aggregator that created the post
104
+
community_did TEXT NOT NULL, -- DID of community post was created in
105
+
post_uri TEXT NOT NULL, -- AT-URI of the post record
106
+
post_cid TEXT NOT NULL, -- CID of the post
107
+
108
+
-- Timestamp
109
+
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- When post was created
110
+
111
+
-- Constraints
112
+
UNIQUE(post_uri), -- Each post tracked once
113
+
CONSTRAINT fk_aggregator_posts_agg FOREIGN KEY (aggregator_did) REFERENCES aggregators(did) ON DELETE CASCADE,
114
+
CONSTRAINT fk_aggregator_posts_comm FOREIGN KEY (community_did) REFERENCES communities(did) ON DELETE CASCADE
115
+
);
116
+
117
+
-- Indexes for rate limiting queries (CRITICAL PATH - used on every aggregator post)
118
+
CREATE INDEX idx_aggregator_posts_rate_limit ON aggregator_posts(aggregator_did, community_did, created_at DESC);
119
+
120
+
-- Indexes for stats
121
+
CREATE INDEX idx_aggregator_posts_agg_did ON aggregator_posts(aggregator_did, created_at DESC);
122
+
CREATE INDEX idx_aggregator_posts_comm_did ON aggregator_posts(community_did, created_at DESC);
123
+
124
+
-- Comments
125
+
COMMENT ON TABLE aggregator_posts IS 'AppView-only tracking of posts created by aggregators for rate limiting and stats';
126
+
COMMENT ON INDEX idx_aggregator_posts_rate_limit IS 'CRITICAL: Fast rate limit checks (posts in last hour per community)';
127
+
128
+
129
+
-- ============================================================================
130
+
-- Trigger: Update aggregator stats when authorizations change
131
+
-- Purpose: Keep aggregators.communities_using count accurate
132
+
-- ============================================================================
133
+
-- +goose StatementBegin
134
+
CREATE OR REPLACE FUNCTION update_aggregator_communities_count()
135
+
RETURNS TRIGGER AS $$
136
+
BEGIN
137
+
-- Recalculate communities_using count for affected aggregator
138
+
IF TG_OP = 'DELETE' THEN
139
+
UPDATE aggregators
140
+
SET communities_using = (
141
+
SELECT COUNT(*)
142
+
FROM aggregator_authorizations
143
+
WHERE aggregator_did = OLD.aggregator_did
144
+
AND enabled = true
145
+
)
146
+
WHERE did = OLD.aggregator_did;
147
+
RETURN OLD;
148
+
ELSE
149
+
UPDATE aggregators
150
+
SET communities_using = (
151
+
SELECT COUNT(*)
152
+
FROM aggregator_authorizations
153
+
WHERE aggregator_did = NEW.aggregator_did
154
+
AND enabled = true
155
+
)
156
+
WHERE did = NEW.aggregator_did;
157
+
RETURN NEW;
158
+
END IF;
159
+
END;
160
+
$$ LANGUAGE plpgsql;
161
+
-- +goose StatementEnd
162
+
163
+
CREATE TRIGGER trigger_update_aggregator_communities_count
164
+
AFTER INSERT OR UPDATE OR DELETE ON aggregator_authorizations
165
+
FOR EACH ROW
166
+
EXECUTE FUNCTION update_aggregator_communities_count();
167
+
168
+
COMMENT ON FUNCTION update_aggregator_communities_count IS 'Maintains aggregators.communities_using count when authorizations change';
169
+
170
+
171
+
-- ============================================================================
172
+
-- Trigger: Update aggregator stats when posts are created
173
+
-- Purpose: Keep aggregators.posts_created count accurate
174
+
-- ============================================================================
175
+
-- +goose StatementBegin
176
+
CREATE OR REPLACE FUNCTION update_aggregator_posts_count()
177
+
RETURNS TRIGGER AS $$
178
+
BEGIN
179
+
IF TG_OP = 'INSERT' THEN
180
+
UPDATE aggregators
181
+
SET posts_created = posts_created + 1
182
+
WHERE did = NEW.aggregator_did;
183
+
RETURN NEW;
184
+
ELSIF TG_OP = 'DELETE' THEN
185
+
UPDATE aggregators
186
+
SET posts_created = posts_created - 1
187
+
WHERE did = OLD.aggregator_did;
188
+
RETURN OLD;
189
+
END IF;
190
+
END;
191
+
$$ LANGUAGE plpgsql;
192
+
-- +goose StatementEnd
193
+
194
+
CREATE TRIGGER trigger_update_aggregator_posts_count
195
+
AFTER INSERT OR DELETE ON aggregator_posts
196
+
FOR EACH ROW
197
+
EXECUTE FUNCTION update_aggregator_posts_count();
198
+
199
+
COMMENT ON FUNCTION update_aggregator_posts_count IS 'Maintains aggregators.posts_created count when posts are tracked';
200
+
201
+
202
+
-- +goose Down
203
+
-- Drop triggers first
204
+
DROP TRIGGER IF EXISTS trigger_update_aggregator_posts_count ON aggregator_posts;
205
+
DROP TRIGGER IF EXISTS trigger_update_aggregator_communities_count ON aggregator_authorizations;
206
+
207
+
-- Drop functions
208
+
DROP FUNCTION IF EXISTS update_aggregator_posts_count();
209
+
DROP FUNCTION IF EXISTS update_aggregator_communities_count();
210
+
211
+
-- Drop tables in reverse order (respects foreign keys)
212
+
DROP TABLE IF EXISTS aggregator_posts CASCADE;
213
+
DROP TABLE IF EXISTS aggregator_authorizations CASCADE;
214
+
DROP TABLE IF EXISTS aggregators CASCADE;