176 lines
9.0 KiB
PL/PgSQL
176 lines
9.0 KiB
PL/PgSQL
BEGIN;
|
|
|
|
CREATE TABLE web_event (
|
|
id bigserial PRIMARY KEY,
|
|
tenant_id uuid NOT NULL REFERENCES tenant(id),
|
|
household_id uuid NOT NULL,
|
|
subject_id uuid,
|
|
event_type varchar(80) NOT NULL CHECK (event_type IN (
|
|
'device.presence.changed',
|
|
'conversation.started','conversation.completed','conversation.failed',
|
|
'story.created','story.revised','media.upload.completed','export.ready',
|
|
'privacy_request.updated','device.command.updated')),
|
|
resource_type varchar(40) NOT NULL,
|
|
resource_public_id uuid NOT NULL,
|
|
status varchar(40) NOT NULL,
|
|
occurred_at timestamptz NOT NULL DEFAULT now(),
|
|
expires_at timestamptz NOT NULL DEFAULT (now() + interval '15 minutes'),
|
|
CONSTRAINT web_event_household_tenant_fk FOREIGN KEY (household_id,tenant_id) REFERENCES household(id,tenant_id),
|
|
CONSTRAINT web_event_subject_tenant_fk FOREIGN KEY (subject_id,tenant_id) REFERENCES subject_profile(id,tenant_id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX web_event_household_cursor_idx ON web_event (tenant_id,household_id,id);
|
|
CREATE INDEX web_event_expiry_idx ON web_event (expires_at);
|
|
|
|
CREATE FUNCTION append_web_event(
|
|
p_tenant uuid,p_household uuid,p_subject uuid,p_event_type varchar,
|
|
p_resource_type varchar,p_resource_public_id uuid,p_status varchar
|
|
) RETURNS void LANGUAGE plpgsql AS $$
|
|
DECLARE new_id bigint;
|
|
BEGIN
|
|
INSERT INTO web_event(tenant_id,household_id,subject_id,event_type,resource_type,resource_public_id,status)
|
|
VALUES(p_tenant,p_household,p_subject,p_event_type,p_resource_type,p_resource_public_id,p_status)
|
|
RETURNING id INTO new_id;
|
|
PERFORM pg_notify('web_event',new_id::text);
|
|
END;
|
|
$$;
|
|
|
|
CREATE TABLE device_presence_state (
|
|
tenant_id uuid NOT NULL REFERENCES tenant(id),
|
|
device_id uuid NOT NULL,
|
|
household_id uuid NOT NULL,
|
|
subject_id uuid NOT NULL,
|
|
online boolean NOT NULL,
|
|
last_seen_at timestamptz NOT NULL,
|
|
changed_at timestamptz NOT NULL,
|
|
PRIMARY KEY (tenant_id,device_id),
|
|
CONSTRAINT device_presence_device_tenant_fk FOREIGN KEY (device_id,tenant_id) REFERENCES device(id,tenant_id),
|
|
CONSTRAINT device_presence_household_tenant_fk FOREIGN KEY (household_id,tenant_id) REFERENCES household(id,tenant_id),
|
|
CONSTRAINT device_presence_subject_tenant_fk FOREIGN KEY (subject_id,tenant_id) REFERENCES subject_profile(id,tenant_id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX device_presence_online_seen_idx ON device_presence_state (online,last_seen_at) WHERE online;
|
|
|
|
CREATE FUNCTION record_device_presence(
|
|
p_tenant uuid,p_device uuid,p_observed_at timestamptz
|
|
) RETURNS void LANGUAGE plpgsql AS $$
|
|
DECLARE binding device_binding%ROWTYPE; device_public uuid; was_online boolean; had_state boolean;
|
|
BEGIN
|
|
SELECT * INTO binding FROM device_binding
|
|
WHERE tenant_id=p_tenant AND device_id=p_device AND unbound_at IS NULL;
|
|
IF NOT FOUND THEN RETURN; END IF;
|
|
SELECT public_id INTO device_public FROM device WHERE id=p_device AND tenant_id=p_tenant;
|
|
SELECT online INTO was_online FROM device_presence_state
|
|
WHERE tenant_id=p_tenant AND device_id=p_device FOR UPDATE;
|
|
had_state := FOUND;
|
|
INSERT INTO device_presence_state(tenant_id,device_id,household_id,subject_id,online,last_seen_at,changed_at)
|
|
VALUES(p_tenant,p_device,binding.household_id,binding.subject_id,true,p_observed_at,p_observed_at)
|
|
ON CONFLICT (tenant_id,device_id) DO UPDATE SET
|
|
household_id=EXCLUDED.household_id,subject_id=EXCLUDED.subject_id,online=true,
|
|
last_seen_at=GREATEST(device_presence_state.last_seen_at,EXCLUDED.last_seen_at),
|
|
changed_at=CASE WHEN device_presence_state.online THEN device_presence_state.changed_at ELSE EXCLUDED.changed_at END;
|
|
IF NOT had_state OR NOT was_online THEN
|
|
PERFORM append_web_event(p_tenant,binding.household_id,binding.subject_id,
|
|
'device.presence.changed','device',device_public,'online');
|
|
END IF;
|
|
END;
|
|
$$;
|
|
|
|
CREATE FUNCTION expire_device_presence(
|
|
p_stale_before timestamptz,p_limit integer
|
|
) RETURNS integer LANGUAGE plpgsql AS $$
|
|
DECLARE item record; expired_count integer := 0;
|
|
BEGIN
|
|
FOR item IN
|
|
SELECT state.tenant_id,state.device_id,state.household_id,state.subject_id,device.public_id
|
|
FROM device_presence_state state
|
|
JOIN device ON device.id=state.device_id AND device.tenant_id=state.tenant_id
|
|
WHERE state.online AND state.last_seen_at<p_stale_before
|
|
ORDER BY state.last_seen_at
|
|
FOR UPDATE OF state SKIP LOCKED LIMIT p_limit
|
|
LOOP
|
|
UPDATE device_presence_state SET online=false,changed_at=now()
|
|
WHERE tenant_id=item.tenant_id AND device_id=item.device_id AND online;
|
|
IF FOUND THEN
|
|
PERFORM append_web_event(item.tenant_id,item.household_id,item.subject_id,
|
|
'device.presence.changed','device',item.public_id,'offline');
|
|
expired_count := expired_count + 1;
|
|
END IF;
|
|
END LOOP;
|
|
RETURN expired_count;
|
|
END;
|
|
$$;
|
|
|
|
CREATE FUNCTION conversation_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE mapped varchar;
|
|
BEGIN
|
|
IF TG_OP='UPDATE' AND NEW.status IS NOT DISTINCT FROM OLD.status THEN RETURN NEW; END IF;
|
|
mapped := CASE NEW.status WHEN 'ACTIVE' THEN 'conversation.started' WHEN 'COMPLETED' THEN 'conversation.completed' WHEN 'FAILED' THEN 'conversation.failed' END;
|
|
IF mapped IS NOT NULL THEN
|
|
PERFORM append_web_event(NEW.tenant_id,NEW.household_id,NEW.subject_id,mapped,'conversation',NEW.public_id,NEW.status);
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER conversation_web_event AFTER INSERT OR UPDATE OF status ON conversation FOR EACH ROW EXECUTE FUNCTION conversation_web_event_trigger();
|
|
|
|
CREATE FUNCTION story_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE household uuid; mapped varchar;
|
|
BEGIN
|
|
IF TG_OP='UPDATE' AND NEW.version IS NOT DISTINCT FROM OLD.version THEN RETURN NEW; END IF;
|
|
SELECT household_id INTO household FROM subject_profile WHERE id=NEW.subject_id AND tenant_id=NEW.tenant_id;
|
|
mapped := CASE WHEN TG_OP='INSERT' THEN 'story.created' ELSE 'story.revised' END;
|
|
PERFORM append_web_event(NEW.tenant_id,household,NEW.subject_id,mapped,'story',NEW.public_id,NEW.status);
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER story_web_event AFTER INSERT OR UPDATE OF version ON story_draft FOR EACH ROW EXECUTE FUNCTION story_web_event_trigger();
|
|
|
|
CREATE FUNCTION media_batch_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE c conversation%ROWTYPE;
|
|
BEGIN
|
|
IF NEW.status='COMPLETED' AND (TG_OP='INSERT' OR OLD.status IS DISTINCT FROM NEW.status) THEN
|
|
SELECT * INTO c FROM conversation WHERE id=NEW.conversation_id AND tenant_id=NEW.tenant_id;
|
|
PERFORM append_web_event(NEW.tenant_id,c.household_id,c.subject_id,'media.upload.completed','upload_batch',NEW.public_id,NEW.status);
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER media_batch_web_event AFTER INSERT OR UPDATE OF status ON upload_batch FOR EACH ROW EXECUTE FUNCTION media_batch_web_event_trigger();
|
|
|
|
CREATE FUNCTION export_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE household uuid;
|
|
BEGIN
|
|
IF NEW.status='READY' AND (TG_OP='INSERT' OR OLD.status IS DISTINCT FROM NEW.status) THEN
|
|
SELECT household_id INTO household FROM subject_profile WHERE id=NEW.subject_id AND tenant_id=NEW.tenant_id;
|
|
PERFORM append_web_event(NEW.tenant_id,household,NEW.subject_id,'export.ready','export_task',NEW.public_id,NEW.status);
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER export_web_event AFTER INSERT OR UPDATE OF status ON export_task FOR EACH ROW EXECUTE FUNCTION export_web_event_trigger();
|
|
|
|
CREATE FUNCTION privacy_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE household uuid;
|
|
BEGIN
|
|
IF TG_OP='UPDATE' AND NEW.status IS NOT DISTINCT FROM OLD.status THEN RETURN NEW; END IF;
|
|
SELECT household_id INTO household FROM subject_profile WHERE id=NEW.subject_id AND tenant_id=NEW.tenant_id;
|
|
PERFORM append_web_event(NEW.tenant_id,household,NEW.subject_id,'privacy_request.updated','privacy_request',NEW.public_id,NEW.status);
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER privacy_web_event AFTER INSERT OR UPDATE OF status ON privacy_request FOR EACH ROW EXECUTE FUNCTION privacy_web_event_trigger();
|
|
|
|
CREATE FUNCTION device_command_web_event_trigger() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
DECLARE binding device_binding%ROWTYPE;
|
|
BEGIN
|
|
IF TG_OP='UPDATE' AND NEW.status IS NOT DISTINCT FROM OLD.status THEN RETURN NEW; END IF;
|
|
SELECT * INTO binding FROM device_binding WHERE device_id=NEW.device_id AND tenant_id=NEW.tenant_id AND unbound_at IS NULL;
|
|
IF FOUND THEN
|
|
PERFORM append_web_event(NEW.tenant_id,binding.household_id,binding.subject_id,'device.command.updated','device_command',NEW.public_id,NEW.status);
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER device_command_web_event AFTER INSERT OR UPDATE OF status ON device_command FOR EACH ROW EXECUTE FUNCTION device_command_web_event_trigger();
|
|
|
|
COMMIT;
|