239 lines
9.5 KiB
SQL
239 lines
9.5 KiB
SQL
-- 0009_private_message_pipeline: second-stage private text message storage.
|
|
--
|
|
-- Large tables are partitioned from the first version of the message module:
|
|
-- message_boxes/dialogs/user_update_events/dispatch_outbox by owner/target user,
|
|
-- private_messages by sender user for random_id idempotency locality.
|
|
|
|
ALTER TABLE IF EXISTS messages RENAME TO messages_legacy;
|
|
ALTER TABLE IF EXISTS dialogs RENAME TO dialogs_legacy;
|
|
|
|
CREATE TABLE IF NOT EXISTS private_messages (
|
|
id BIGINT GENERATED BY DEFAULT AS IDENTITY,
|
|
sender_user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
recipient_user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
random_id BIGINT NOT NULL DEFAULT 0,
|
|
message_date INT NOT NULL,
|
|
body TEXT NOT NULL DEFAULT '',
|
|
entities JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (sender_user_id, id),
|
|
CONSTRAINT private_messages_nonempty_body CHECK (body <> '')
|
|
) PARTITION BY HASH (sender_user_id);
|
|
|
|
CREATE UNIQUE INDEX IF NOT EXISTS private_messages_sender_random_idx
|
|
ON private_messages (sender_user_id, random_id)
|
|
WHERE random_id <> 0;
|
|
CREATE INDEX IF NOT EXISTS private_messages_recipient_date_idx
|
|
ON private_messages (recipient_user_id, message_date DESC, id DESC);
|
|
|
|
DO $$
|
|
DECLARE
|
|
i int;
|
|
BEGIN
|
|
FOR i IN 0..63 LOOP
|
|
EXECUTE format(
|
|
'CREATE TABLE IF NOT EXISTS private_messages_p%s PARTITION OF private_messages FOR VALUES WITH (MODULUS 64, REMAINDER %s)',
|
|
lpad(i::text, 2, '0'),
|
|
i
|
|
);
|
|
END LOOP;
|
|
END $$;
|
|
|
|
CREATE TABLE IF NOT EXISTS message_boxes (
|
|
owner_user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
box_id INT NOT NULL,
|
|
private_message_id BIGINT NOT NULL,
|
|
message_sender_id BIGINT NOT NULL,
|
|
peer_type VARCHAR(16) NOT NULL,
|
|
peer_id BIGINT NOT NULL,
|
|
from_user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE RESTRICT,
|
|
message_date INT NOT NULL,
|
|
outgoing BOOLEAN NOT NULL DEFAULT false,
|
|
body TEXT NOT NULL DEFAULT '',
|
|
entities JSONB NOT NULL DEFAULT '[]'::jsonb,
|
|
pts INT NOT NULL DEFAULT 0,
|
|
deleted BOOLEAN NOT NULL DEFAULT false,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (owner_user_id, box_id),
|
|
UNIQUE (owner_user_id, private_message_id),
|
|
CONSTRAINT message_boxes_peer_type_check CHECK (peer_type IN ('user')),
|
|
FOREIGN KEY (message_sender_id, private_message_id)
|
|
REFERENCES private_messages(sender_user_id, id) ON DELETE CASCADE
|
|
) PARTITION BY HASH (owner_user_id);
|
|
|
|
CREATE INDEX IF NOT EXISTS message_boxes_dialog_seek_idx
|
|
ON message_boxes (owner_user_id, peer_type, peer_id, box_id DESC)
|
|
WHERE NOT deleted;
|
|
CREATE INDEX IF NOT EXISTS message_boxes_owner_date_idx
|
|
ON message_boxes (owner_user_id, message_date DESC, box_id DESC)
|
|
WHERE NOT deleted;
|
|
CREATE INDEX IF NOT EXISTS message_boxes_private_lookup_idx
|
|
ON message_boxes (private_message_id, owner_user_id);
|
|
|
|
DO $$
|
|
DECLARE
|
|
i int;
|
|
BEGIN
|
|
FOR i IN 0..63 LOOP
|
|
EXECUTE format(
|
|
'CREATE TABLE IF NOT EXISTS message_boxes_p%s PARTITION OF message_boxes FOR VALUES WITH (MODULUS 64, REMAINDER %s)',
|
|
lpad(i::text, 2, '0'),
|
|
i
|
|
);
|
|
END LOOP;
|
|
END $$;
|
|
|
|
CREATE TABLE IF NOT EXISTS dialogs (
|
|
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
peer_type VARCHAR(16) NOT NULL,
|
|
peer_id BIGINT NOT NULL,
|
|
top_message_id INT NOT NULL DEFAULT 0,
|
|
top_message_date INT NOT NULL DEFAULT 0,
|
|
read_inbox_max_id INT NOT NULL DEFAULT 0,
|
|
read_outbox_max_id INT NOT NULL DEFAULT 0,
|
|
unread_count INT NOT NULL DEFAULT 0,
|
|
unread_mentions_count INT NOT NULL DEFAULT 0,
|
|
unread_reactions_count INT NOT NULL DEFAULT 0,
|
|
pinned BOOLEAN NOT NULL DEFAULT false,
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (user_id, peer_type, peer_id),
|
|
CONSTRAINT dialogs_peer_type_check CHECK (peer_type IN ('user'))
|
|
) PARTITION BY HASH (user_id);
|
|
|
|
CREATE INDEX IF NOT EXISTS dialogs_user_top_message_idx
|
|
ON dialogs (user_id, pinned DESC, top_message_date DESC, top_message_id DESC, peer_id DESC);
|
|
CREATE INDEX IF NOT EXISTS dialogs_user_pinned_idx
|
|
ON dialogs (user_id, pinned) WHERE pinned;
|
|
|
|
DO $$
|
|
DECLARE
|
|
i int;
|
|
BEGIN
|
|
FOR i IN 0..63 LOOP
|
|
EXECUTE format(
|
|
'CREATE TABLE IF NOT EXISTS dialogs_p%s PARTITION OF dialogs FOR VALUES WITH (MODULUS 64, REMAINDER %s)',
|
|
lpad(i::text, 2, '0'),
|
|
i
|
|
);
|
|
END LOOP;
|
|
END $$;
|
|
|
|
CREATE TABLE IF NOT EXISTS user_update_events (
|
|
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
pts INT NOT NULL,
|
|
pts_count INT NOT NULL DEFAULT 1,
|
|
date INT NOT NULL,
|
|
event_type VARCHAR(32) NOT NULL,
|
|
message_box_id INT,
|
|
peer_type VARCHAR(16),
|
|
peer_id BIGINT,
|
|
max_id INT NOT NULL DEFAULT 0,
|
|
still_unread_count INT NOT NULL DEFAULT 0,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (user_id, pts),
|
|
CONSTRAINT user_update_events_type_check CHECK (event_type IN ('new_message', 'read_history_inbox', 'noop')),
|
|
CONSTRAINT user_update_events_peer_type_check CHECK (peer_type IS NULL OR peer_type IN ('user')),
|
|
FOREIGN KEY (user_id, message_box_id) REFERENCES message_boxes(owner_user_id, box_id) ON DELETE CASCADE
|
|
) PARTITION BY HASH (user_id);
|
|
|
|
DO $$
|
|
DECLARE
|
|
i int;
|
|
BEGIN
|
|
FOR i IN 0..63 LOOP
|
|
EXECUTE format(
|
|
'CREATE TABLE IF NOT EXISTS user_update_events_p%s PARTITION OF user_update_events FOR VALUES WITH (MODULUS 64, REMAINDER %s)',
|
|
lpad(i::text, 2, '0'),
|
|
i
|
|
);
|
|
END LOOP;
|
|
END $$;
|
|
|
|
CREATE TABLE IF NOT EXISTS dispatch_outbox (
|
|
id BIGINT GENERATED BY DEFAULT AS IDENTITY,
|
|
target_user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
pts INT NOT NULL,
|
|
event_type VARCHAR(32) NOT NULL,
|
|
exclude_session_id BIGINT NOT NULL DEFAULT 0,
|
|
status VARCHAR(16) NOT NULL DEFAULT 'pending',
|
|
attempts INT NOT NULL DEFAULT 0,
|
|
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
last_error TEXT NOT NULL DEFAULT '',
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (target_user_id, id),
|
|
CONSTRAINT dispatch_outbox_status_check CHECK (status IN ('pending', 'dispatching', 'delivered', 'failed')),
|
|
FOREIGN KEY (target_user_id, pts) REFERENCES user_update_events(user_id, pts) ON DELETE CASCADE
|
|
) PARTITION BY HASH (target_user_id);
|
|
|
|
CREATE INDEX IF NOT EXISTS dispatch_outbox_pending_idx
|
|
ON dispatch_outbox (status, next_attempt_at, target_user_id, id)
|
|
WHERE status = 'pending';
|
|
|
|
DO $$
|
|
DECLARE
|
|
i int;
|
|
BEGIN
|
|
FOR i IN 0..63 LOOP
|
|
EXECUTE format(
|
|
'CREATE TABLE IF NOT EXISTS dispatch_outbox_p%s PARTITION OF dispatch_outbox FOR VALUES WITH (MODULUS 64, REMAINDER %s)',
|
|
lpad(i::text, 2, '0'),
|
|
i
|
|
);
|
|
END LOOP;
|
|
END $$;
|
|
|
|
ALTER TABLE update_states
|
|
ADD COLUMN IF NOT EXISTS user_id BIGINT NOT NULL DEFAULT 0;
|
|
ALTER TABLE update_states DROP CONSTRAINT IF EXISTS update_states_pkey;
|
|
ALTER TABLE update_states ADD PRIMARY KEY (auth_key_id, user_id);
|
|
CREATE INDEX IF NOT EXISTS update_states_user_id_idx ON update_states (user_id);
|
|
|
|
DO $$
|
|
DECLARE
|
|
r record;
|
|
private_id bigint;
|
|
BEGIN
|
|
IF to_regclass('messages_legacy') IS NULL THEN
|
|
RETURN;
|
|
END IF;
|
|
|
|
FOR r IN
|
|
SELECT id, owner_user_id, peer_type, peer_id, from_user_id, message_date, outgoing, body, entities
|
|
FROM messages_legacy
|
|
ORDER BY owner_user_id, id
|
|
LOOP
|
|
INSERT INTO private_messages (
|
|
sender_user_id, recipient_user_id, random_id, message_date, body, entities
|
|
) VALUES (
|
|
r.from_user_id,
|
|
r.owner_user_id,
|
|
0,
|
|
r.message_date,
|
|
r.body,
|
|
r.entities
|
|
)
|
|
RETURNING id INTO private_id;
|
|
|
|
INSERT INTO message_boxes (
|
|
owner_user_id, box_id, private_message_id, message_sender_id, peer_type, peer_id,
|
|
from_user_id, message_date, outgoing, body, entities
|
|
) VALUES (
|
|
r.owner_user_id, r.id, private_id, r.from_user_id, r.peer_type, r.peer_id,
|
|
r.from_user_id, r.message_date, r.outgoing, r.body, r.entities
|
|
)
|
|
ON CONFLICT (owner_user_id, box_id) DO NOTHING;
|
|
END LOOP;
|
|
END $$;
|
|
|
|
INSERT INTO dialogs (
|
|
user_id, peer_type, peer_id, top_message_id, top_message_date,
|
|
read_inbox_max_id, read_outbox_max_id, unread_count,
|
|
unread_mentions_count, unread_reactions_count, pinned, updated_at
|
|
)
|
|
SELECT
|
|
user_id, peer_type, peer_id, top_message_id, top_message_date,
|
|
read_inbox_max_id, read_outbox_max_id, unread_count,
|
|
unread_mentions_count, unread_reactions_count, pinned, updated_at
|
|
FROM dialogs_legacy
|
|
ON CONFLICT (user_id, peer_type, peer_id) DO NOTHING;
|