Drop raw visits

migrations/000056_create_peer_logs_table.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+DROP TRIGGER IF EXISTS on_peer_update ON peers;
+DROP FUNCTION IF EXISTS insert_peer_log;
+DROP TABLE IF EXISTS peer_logs;
+COMMIT;

migrations/000056_create_peer_logs_table.up.sql

@@ -0,0 +1,41 @@
+BEGIN;
+
+CREATE TABLE peer_logs
+(
+    id         INT GENERATED ALWAYS AS IDENTITY,
+    peer_id    INT         NOT NULL,
+    field      TEXT        NOT NULL,
+    old        TEXT        NOT NULL,
+    new        TEXT        NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+
+    CONSTRAINT fk_peer_logs_peer_id FOREIGN KEY (peer_id) REFERENCES peers (id),
+
+    PRIMARY KEY (id)
+);
+
+CREATE OR REPLACE FUNCTION insert_peer_log()
+    RETURNS TRIGGER AS
+$$
+BEGIN
+    IF OLD.agent_version_id != NEW.agent_version_id THEN
+        INSERT INTO peer_logs (peer_id, field, old, new, created_at)
+        VALUES (NEW.id, 'agent_version_id', OLD.agent_version_id, NEW.agent_version_id, NOW());
+        RETURN NEW;
+    END IF;
+
+    IF OLD.protocols_set_id != NEW.protocols_set_id THEN
+        INSERT INTO peer_logs (peer_id, field, old, new, created_at)
+        VALUES (NEW.id, 'protocols_set_id', OLD.protocols_set_id, NEW.protocols_set_id, NOW());
+        RETURN NEW;
+    END IF;
+END;
+$$ LANGUAGE 'plpgsql';
+
+CREATE TRIGGER on_peer_update
+    BEFORE UPDATE
+    ON peers
+    FOR EACH ROW
+EXECUTE PROCEDURE insert_peer_log();
+
+COMMIT;

migrations/000057_drop_pegasys_tables.down.sql

@@ -0,0 +1,27 @@
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS pegasys_neighbours
+(
+    id                SERIAL,
+    peer_id           VARCHAR(100) NOT NULL,
+    neighbour_peer_id VARCHAR(100) NOT NULL,
+    created_at        TIMESTAMPTZ,
+    crawl_start_at    TIMESTAMPTZ,
+
+    PRIMARY KEY (id)
+);
+
+
+CREATE TABLE IF NOT EXISTS pegasys_connections
+(
+    id           SERIAL,
+    peer_id      VARCHAR(100) NOT NULL,
+    dial_attempt TIMESTAMPTZ,
+    latency      INTERVAL,
+    is_succeed   BOOLEAN,
+    error        VARCHAR(100),
+
+    PRIMARY KEY (id)
+);
+
+COMMIT;

migrations/000057_drop_pegasys_tables.up.sql

@@ -0,0 +1,4 @@
+BEGIN;
+DROP TABLE pegasys_connections;
+DROP TABLE pegasys_neighbours;
+COMMIT;

migrations/000058_add_upsert_session_function.down.sql

@@ -0,0 +1 @@
+DROP FUNCTION IF EXISTS upsert_session;

migrations/000058_add_upsert_session_function.up.sql

@@ -0,0 +1,60 @@
+CREATE OR REPLACE FUNCTION upsert_session(
+    visit_peer_id INT,
+    new_visit_started_at TIMESTAMPTZ,
+    new_visit_ended_at TIMESTAMPTZ,
+    new_error dial_error
+)
+    RETURNS INT AS
+$$
+DECLARE
+    upserted_session_id INT;
+BEGIN
+    IF new_error IS NULL THEN
+        INSERT INTO sessions (peer_id,
+                              first_successful_dial,
+                              last_successful_dial,
+                              first_failed_dial,
+                              next_dial_attempt,
+                              successful_dials,
+                              finished,
+                              created_at,
+                              updated_at)
+        VALUES (visit_peer_id, new_visit_ended_at, new_visit_ended_at, '1970-01-01',
+                new_visit_ended_at + '30s'::interval, 1, false, NOW(), NOW())
+        ON CONFLICT ON CONSTRAINT uq_peer_id_first_failed_dial DO UPDATE
+            SET last_successful_dial = EXCLUDED.last_successful_dial,
+                successful_dials     = sessions.successful_dials + 1,
+                updated_at           = EXCLUDED.updated_at,
+                next_dial_attempt    =
+                    CASE
+                        WHEN 0.5 *
+                             (EXCLUDED.last_successful_dial - sessions.first_successful_dial) <
+                             '30s'::interval THEN
+                            EXCLUDED.last_successful_dial + '30s'::interval
+                        WHEN 0.5 *
+                             (EXCLUDED.last_successful_dial - sessions.first_successful_dial) >
+                             '15m'::interval THEN
+                            EXCLUDED.last_successful_dial + '15m'::interval
+                        ELSE
+                                EXCLUDED.last_successful_dial +
+                                0.5 *
+                                (EXCLUDED.last_successful_dial - sessions.first_successful_dial)
+                        END
+        RETURNING id INTO upserted_session_id;
+    ELSE
+        UPDATE sessions
+        SET first_failed_dial = new_visit_started_at,
+            min_duration      = last_successful_dial - first_successful_dial,
+            max_duration      = new_visit_started_at - first_successful_dial,
+            finished          = true,
+            updated_at        = NOW(),
+            next_dial_attempt = null,
+            finish_reason     = new_error
+        WHERE peer_id = visit_peer_id
+          AND finished = false
+        RETURNING id INTO upserted_session_id;
+    END IF;
+
+    RETURN upserted_session_id;
+END;
+$$ LANGUAGE 'plpgsql';

migrations/000059_update_upsert_peer_function.down.sql

@@ -0,0 +1,39 @@
+BEGIN;
+
+CREATE OR REPLACE FUNCTION upsert_peer(
+    new_multi_hash TEXT,
+    new_agent_version_id INT,
+    new_protocols_set_id INT,
+    new_created_at TIMESTAMPTZ DEFAULT NOW()
+) RETURNS INT AS
+$upsert_peer$
+DECLARE
+    peer_id INT;
+    peer    peers%rowtype;
+BEGIN
+    SELECT *
+    FROM peers p
+    WHERE p.multi_hash = new_multi_hash
+    INTO peer;
+
+    IF peer IS NULL THEN
+        INSERT INTO peers (multi_hash, agent_version_id, protocols_set_id, updated_at, created_at)
+        VALUES (new_multi_hash, new_agent_version_id, new_protocols_set_id, new_created_at, new_created_at)
+        RETURNING id INTO peer_id;
+
+        RETURN peer_id;
+    END IF;
+
+    IF peer.agent_version_id != new_agent_version_id OR peer.protocols_set_id != new_protocols_set_id THEN
+        UPDATE peers
+        SET agent_version_id = new_agent_version_id,
+            protocols_set_id = new_protocols_set_id,
+            updated_at       = new_created_at
+        WHERE id = peer.id;
+    END IF;
+
+    RETURN peer.id;
+END;
+$upsert_peer$ LANGUAGE plpgsql;
+
+COMMIT;

migrations/000059_update_upsert_peer_function.up.sql

@@ -0,0 +1,45 @@
+BEGIN;
+
+CREATE OR REPLACE FUNCTION upsert_peer(
+    new_multi_hash TEXT,
+    new_agent_version_id INT,
+    new_protocols_set_id INT,
+    new_created_at TIMESTAMPTZ DEFAULT NOW()
+) RETURNS INT AS
+$upsert_peer$
+DECLARE
+    peer_id INT;
+    peer    peers%rowtype;
+BEGIN
+    SELECT *
+    FROM peers p
+    WHERE p.multi_hash = new_multi_hash
+    INTO peer;
+
+    IF peer IS NULL THEN
+        INSERT INTO peers (multi_hash, agent_version_id, protocols_set_id, updated_at, created_at)
+        VALUES (new_multi_hash, new_agent_version_id, new_protocols_set_id, new_created_at, new_created_at)
+        RETURNING id INTO peer_id;
+
+        RETURN peer_id;
+    END IF;
+
+    IF new_agent_version_id IS NOT NULL AND new_agent_version_id != peer.agent_version_id THEN
+        UPDATE peers
+        SET agent_version_id = new_agent_version_id,
+            updated_at       = new_created_at
+        WHERE id = peer.id;
+    END IF;
+
+    IF new_protocols_set_id IS NOT NULL AND new_protocols_set_id != peer.protocols_set_id THEN
+        UPDATE peers
+        SET protocols_set_id = new_protocols_set_id,
+            updated_at       = new_created_at
+        WHERE id = peer.id;
+    END IF;
+
+    RETURN peer.id;
+END;
+$upsert_peer$ LANGUAGE plpgsql;
+
+COMMIT;

migrations/000060_add_upsert_multi_addresses_function.down.sql

@@ -0,0 +1 @@
+DROP FUNCTION IF EXISTS upsert_multi_addresses;

migrations/000060_add_upsert_multi_addresses_function.up.sql

@@ -0,0 +1,35 @@
+CREATE OR REPLACE FUNCTION upsert_multi_addresses(
+    new_multi_addresses TEXT[],
+    new_created_at TIMESTAMPTZ DEFAULT NOW()
+) RETURNS INT[] AS
+$upsert_multi_addresses$
+DECLARE
+    upserted_multi_address_ids INT[];
+BEGIN
+    IF new_multi_addresses IS NULL OR array_length(new_multi_addresses, 1) IS NULL THEN
+        RETURN NULL;
+    END IF;
+
+    WITH insert_multi_addresses AS (
+        SELECT new_multi_addresses_table new_multi_address
+        FROM unnest(new_multi_addresses) new_multi_addresses_table
+                 LEFT JOIN multi_addresses ma ON ma.maddr = new_multi_addresses_table
+        WHERE ma.id IS NULL)
+    INSERT
+    INTO multi_addresses (maddr, updated_at, created_at)
+    SELECT new_multi_address,
+           new_created_at,
+           new_created_at
+    FROM insert_multi_addresses
+    ORDER BY new_multi_address
+    ON CONFLICT DO NOTHING;
+
+    SELECT sort(array_agg(id))
+    FROM multi_addresses
+    WHERE maddr = ANY (new_multi_addresses)
+    INTO upserted_multi_address_ids;
+
+    RETURN upserted_multi_address_ids;
+END;
+$upsert_multi_addresses$ LANGUAGE plpgsql;
+

migrations/000061_add_upsert_multi_address_set_id_function.up.sql

@@ -0,0 +1,45 @@
+BEGIN;
+
+CREATE OR REPLACE FUNCTION upsert_multi_addresses_set_id(
+    new_multi_address_ids INT[]
+) RETURNS INT AS
+$upsert_multi_addresses_set_id$
+DECLARE
+    insert_multi_address_set_ids    INT[];
+    upserted_multi_addresses_set_id INT;
+BEGIN
+    -- filter duplicates and nulls from array and sort it ID ascending
+    SELECT array(SELECT DISTINCT unnest(new_multi_address_ids) WHERE 1 IS NOT NULL ORDER BY 1)
+    INTO insert_multi_address_set_ids;
+
+    IF insert_multi_address_set_ids IS NULL OR array_length(insert_multi_address_set_ids, 1) IS NULL THEN
+        RETURN NULL;
+    END IF;
+
+    -- Check if set of multi_address IDs already exists
+    SELECT id
+    FROM multi_addresses_sets ps
+    WHERE ps.multi_address_ids = insert_multi_address_set_ids
+    INTO upserted_multi_addresses_set_id;
+
+    -- If the set of multi_address IDs does not exist update it
+    IF upserted_multi_addresses_set_id IS NULL THEN
+        INSERT
+        INTO multi_addresses_sets (multi_address_ids, updated_at, created_at) (SELECT insert_multi_address_set_ids, NOW(), NOW()
+                                                                               WHERE insert_multi_address_set_ids IS NOT NULL)
+        ON CONFLICT DO NOTHING
+        RETURNING id INTO upserted_multi_addresses_set_id;
+
+        IF upserted_multi_addresses_set_id IS NULL THEN
+            SELECT id
+            FROM multi_addresses_sets ps
+            WHERE ps.multi_address_ids = insert_multi_address_set_ids
+            INTO upserted_multi_addresses_set_id;
+        END IF;
+    END IF;
+
+    RETURN upserted_multi_addresses_set_id;
+END;
+$upsert_multi_addresses_set_id$ LANGUAGE plpgsql;
+
+COMMIT;