= Напредна тема: Пропагација на податоци помеѓу централна и локална база =

== Опис на напредната тема ==

Како напредна тема во рамки на проектот '''SCS - Smart City Security''' е имплементиран механизам за пропагација на податоци помеѓу централна и локална база на податоци.

Идејата на оваа имплементација е да се симулира Smart City Security систем во кој постои централна база, која ги содржи главните податоци за системот, и локална база, која добива копија од релевантните податоци.

Во ваков систем централната база може да се користи на ниво на град или институција, додека локалната база може да припаѓа на одредена полициска станица, зона или организациски дел од системот.

Целта на пропагацијата е при внесување нови податоци во централната база, тие контролирано да се префрлат и во локалната база.

== Користени табели ==

За демонстрација на пропагацијата се користат следните табели:

* '''gragjanin'''
* '''vozilo'''
* '''kamera'''
* '''prekrsok'''
* '''kazna'''

Во централната база дополнително е креирана табелата:

* '''propagation_log'''

Оваа табела служи за евиденција на сите промени што треба да се префрлат од централната во локалната база.

== Табела propagation_log ==

Табелата '''propagation_log''' претставува лог табела во која се запишуваат сите нови записи што треба да се пропагираат.

Секој запис во оваа табела содржи информации за:

* табелата во која е направена промената
* типот на операција
* идентификаторот на записот
* payload со вредностите на записот
* статус на обработка
* датум на креирање
* датум на обработка
* порака за грешка, доколку пропагацијата не успее

{{{
CREATE TABLE propagation_log
(
log_id           SERIAL PRIMARY KEY,
tabela           VARCHAR(50) NOT NULL,
operacija        VARCHAR(20) NOT NULL,
zapis_id         VARCHAR(50) NOT NULL,
payload          JSONB NOT NULL,
status           VARCHAR(20) DEFAULT 'PENDING',
datum_kreiranje  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
datum_obrabotka  TIMESTAMP,
poraka_greska    TEXT
);
}}}

Статусот може да има една од следните вредности:

* '''PENDING''' - записот чека да биде префрлен во локалната база
* '''DONE''' - записот е успешно префрлен
* '''FAILED''' - настанала грешка при префрлањето

== Функција за логирање на промени ==

За евидентирање на новите записи е креирана функцијата:

{{{
fn_log_propagation_insert()
}}}

Оваа функција се повикува преку trigger-и. Таа го зема новиот внесен ред преку '''NEW''', го претвора во JSONB формат и го запишува во '''propagation_log'''.

{{{
CREATE OR REPLACE FUNCTION fn_log_propagation_insert()
RETURNS TRIGGER AS $$
DECLARE
pk_column TEXT;
pk_value  TEXT;
BEGIN
pk_column := TG_ARGV[0];
pk_value := to_jsonb(NEW) ->> pk_column;

```
INSERT INTO propagation_log (
    tabela,
    operacija,
    zapis_id,
    payload
)
VALUES (
    TG_TABLE_NAME,
    'INSERT',
    pk_value,
    to_jsonb(NEW)
);

RETURN NEW;
```

END;
$$ LANGUAGE plpgsql;
}}}

Функцијата како аргумент го добива името на primary key колоната за табелата на која се однесува trigger-от. На тој начин истата функција може да се користи за повеќе табели.

Користењето на JSONB овозможува целиот запис да се зачува како payload, без потреба да се креира посебна log табела за секоја табела.

== Trigger-и ==

Во централната база се креирани trigger-и за табелите:

* '''gragjanin'''
* '''vozilo'''
* '''kamera'''
* '''prekrsok'''
* '''kazna'''

Trigger-ите се активираат по секој INSERT во соодветната табела. Тие не ги префрлаат директно податоците во локалната база, туку автоматски додаваат запис во '''propagation_log''' со статус '''PENDING'''.

{{{
DROP TRIGGER IF EXISTS trg_log_gragjanin_insert ON gragjanin;
DROP TRIGGER IF EXISTS trg_log_vozilo_insert ON vozilo;
DROP TRIGGER IF EXISTS trg_log_kamera_insert ON kamera;
DROP TRIGGER IF EXISTS trg_log_prekrsok_insert ON prekrsok;
DROP TRIGGER IF EXISTS trg_log_kazna_insert ON kazna;

CREATE TRIGGER trg_log_gragjanin_insert
AFTER INSERT ON gragjanin
FOR EACH ROW
EXECUTE FUNCTION fn_log_propagation_insert('embg');

CREATE TRIGGER trg_log_vozilo_insert
AFTER INSERT ON vozilo
FOR EACH ROW
EXECUTE FUNCTION fn_log_propagation_insert('registarska_oznaka');

CREATE TRIGGER trg_log_kamera_insert
AFTER INSERT ON kamera
FOR EACH ROW
EXECUTE FUNCTION fn_log_propagation_insert('kamera_id');

CREATE TRIGGER trg_log_prekrsok_insert
AFTER INSERT ON prekrsok
FOR EACH ROW
EXECUTE FUNCTION fn_log_propagation_insert('prekrsok_id');

CREATE TRIGGER trg_log_kazna_insert
AFTER INSERT ON kazna
FOR EACH ROW
EXECUTE FUNCTION fn_log_propagation_insert('kazna_id');
}}}

Со ова при секој нов INSERT во некоја од наведените табели автоматски се креира запис во '''propagation_log'''.

== Процедура за пропагација ==

За префрлање на податоците од централната во локалната база е креирана процедурата:

{{{
sp_propagiraj_vo_lokalna_baza()
}}}

Процедурата ги зема сите записи од '''propagation_log''' со статус '''PENDING'''. За секој запис проверува од која табела доаѓа, го чита payload-от и го внесува соодветниот запис во локалната база.

Ако внесувањето е успешно, статусот во '''propagation_log''' се менува во '''DONE'''. Ако настане грешка, статусот се менува во '''FAILED''' и се запишува пораката за грешка.

{{{
CREATE OR REPLACE PROCEDURE sp_propagiraj_vo_lokalna_baza()
LANGUAGE plpgsql
AS $$
DECLARE
r RECORD;
p JSONB;
conn TEXT := 'host=127.0.0.1 port=5430 dbname=scs_local user=postgres password=postgres';
BEGIN
FOR r IN
SELECT *
FROM propagation_log
WHERE status = 'PENDING'
ORDER BY log_id
LOOP
BEGIN
p := r.payload;

```
        IF r.tabela = 'gragjanin' THEN

            PERFORM dblink_exec(
                conn,
                format(
                    'INSERT INTO gragjanin
                     (embg, ime, prezime, adresa, grad, telefonski_broj, datum_ragjanje, pol)
                     VALUES (%L, %L, %L, %L, %L, %L, %L, %L)
                     ON CONFLICT (embg) DO NOTHING',
                    p ->> 'embg',
                    p ->> 'ime',
                    p ->> 'prezime',
                    p ->> 'adresa',
                    p ->> 'grad',
                    p ->> 'telefonski_broj',
                    p ->> 'datum_ragjanje',
                    p ->> 'pol'
                )
            );

        ELSIF r.tabela = 'vozilo' THEN

            PERFORM dblink_exec(
                conn,
                format(
                    'INSERT INTO vozilo
                     (registarska_oznaka, marka, model, boja, embg_sopstvenik)
                     VALUES (%L, %L, %L, %L, %L)
                     ON CONFLICT (registarska_oznaka) DO NOTHING',
                    p ->> 'registarska_oznaka',
                    p ->> 'marka',
                    p ->> 'model',
                    p ->> 'boja',
                    p ->> 'embg_sopstvenik'
                )
            );

        ELSIF r.tabela = 'kamera' THEN

            PERFORM dblink_exec(
                conn,
                format(
                    'INSERT INTO kamera
                     (kamera_id, lokacija, aktivna)
                     VALUES (%L, %L, %L)
                     ON CONFLICT (kamera_id) DO NOTHING',
                    p ->> 'kamera_id',
                    p ->> 'lokacija',
                    p ->> 'aktivna'
                )
            );

        ELSIF r.tabela = 'prekrsok' THEN

            PERFORM dblink_exec(
                conn,
                format(
                    'INSERT INTO prekrsok
                     (prekrsok_id, datum, opis, kamera_id, registarska_oznaka, embg_storitel)
                     VALUES (%L, %L, %L, %L, %L, %L)
                     ON CONFLICT (prekrsok_id) DO NOTHING',
                    p ->> 'prekrsok_id',
                    p ->> 'datum',
                    p ->> 'opis',
                    p ->> 'kamera_id',
                    p ->> 'registarska_oznaka',
                    p ->> 'embg_storitel'
                )
            );

        ELSIF r.tabela = 'kazna' THEN

            PERFORM dblink_exec(
                conn,
                format(
                    'INSERT INTO kazna
                     (kazna_id, prekrsok_id, datum, iznos_za_plakanje, rok_na_plakanje, status)
                     VALUES (%L, %L, %L, %L, %L, %L)
                     ON CONFLICT (kazna_id) DO NOTHING',
                    p ->> 'kazna_id',
                    p ->> 'prekrsok_id',
                    p ->> 'datum',
                    p ->> 'iznos_za_plakanje',
                    p ->> 'rok_na_plakanje',
                    p ->> 'status'
                )
            );

        END IF;

        UPDATE propagation_log
        SET status = 'DONE',
            datum_obrabotka = CURRENT_TIMESTAMP,
            poraka_greska = NULL
        WHERE log_id = r.log_id;

    EXCEPTION WHEN OTHERS THEN
        UPDATE propagation_log
        SET status = 'FAILED',
            datum_obrabotka = CURRENT_TIMESTAMP,
            poraka_greska = SQLERRM
        WHERE log_id = r.log_id;
    END;
END LOOP;
```

END;
$$;
}}}

== Користење на dblink ==

Бидејќи централната и локалната база се две одделни PostgreSQL бази, за комуникација помеѓу нив е користен PostgreSQL extension:

{{{
dblink
}}}

Овој extension овозможува од една база да се извршуваат SQL команди кон друга база.

{{{
CREATE EXTENSION IF NOT EXISTS dblink;
}}}

Во процедурата '''sp_propagiraj_vo_lokalna_baza()''' преку '''dblink_exec''' се извршуваат INSERT команди кон локалната база.

== Тек на пропагацијата ==

Целиот процес се одвива во следните чекори:

1. Се внесува нов запис во некоја од табелите во централната база.
2. Trigger автоматски се активира по INSERT операцијата.
3. Trigger-от ја повикува функцијата '''fn_log_propagation_insert()'''.
4. Во '''propagation_log''' се додава запис со статус '''PENDING'''.
5. Се повикува процедурата '''sp_propagiraj_vo_lokalna_baza()'''.
6. Процедурата ги префрла податоците во локалната база.
7. Статусот во '''propagation_log''' се менува во '''DONE'''.

== Пример за проверка ==

За проверка на записите во лог табелата се користи следното барање:

{{{
SELECT
log_id,
tabela,
operacija,
zapis_id,
status,
datum_kreiranje,
datum_obrabotka,
poraka_greska
FROM propagation_log
ORDER BY log_id;
}}}

По успешна пропагација, записите во колоната '''status''' треба да имаат вредност '''DONE'''.

За проверка дека податоците се префрлени во локалната база се прави SELECT од соодветните табели:

{{{
SELECT * FROM gragjanin;
SELECT * FROM vozilo;
SELECT * FROM kamera;
SELECT * FROM prekrsok;
SELECT * FROM kazna;
}}}

== Заклучок ==

Со оваа напредна тема е имплементиран механизам за пропагација на податоци помеѓу централна и локална база. Решението користи trigger-и, log табела, stored procedure и dblink extension.

Овој пристап е соодветен за Smart City Security систем бидејќи овозможува централно внесување на податоци и нивна контролирана синхронизација кон локални бази, што е применливо во системи каде различни институции или зони треба да имаат пристап до релевантни податоци.