hub3: initial version of postgresql service.

ikuzo/storage/x/postgresql/db.go

@@ -0,0 +1,48 @@
+package postgresql
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	// import the sql.DB driver for postgresql
+	_ "github.com/lib/pq"
+)
+
+type Config struct {
+	DSN          string
+	MaxOpenConns int
+	MaxIdleConns int
+	MaxIdleTime  string
+}
+
+func OpenDB(cfg Config) (*sql.DB, error) {
+	db, err := sql.Open("postgres", cfg.DSN)
+	if err != nil {
+		return nil, err
+	}
+
+	db.SetMaxOpenConns(cfg.MaxOpenConns)
+	db.SetMaxIdleConns(cfg.MaxIdleConns)
+
+	duration, err := time.ParseDuration(cfg.MaxIdleTime)
+	if err != nil {
+		return nil, err
+	}
+
+	db.SetConnMaxIdleTime(duration)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err = db.PingContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := EnsureSchema(cfg.DSN); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}

ikuzo/storage/x/postgresql/migrate.go

@@ -0,0 +1,34 @@
+package postgresql
+
+import (
+	"embed"
+	"fmt"
+
+	"github.com/golang-migrate/migrate/v4"
+	_ "github.com/golang-migrate/migrate/v4/database/postgres"
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+)
+
+//go:embed migrations
+var migrations embed.FS
+
+const schemaVersion = 1
+
+func EnsureSchema(dsn string) error {
+	sourceInstance, err := iofs.New(migrations, "migrations")
+	if err != nil {
+		return fmt.Errorf("invalid source instance, %w", err)
+	}
+
+	m, err := migrate.NewWithSourceInstance("iofs", sourceInstance, dsn)
+	if err != nil {
+		return fmt.Errorf("failed to initialize migrate instance, %w", err)
+	}
+
+	err = m.Migrate(schemaVersion)
+	if err != nil && err != migrate.ErrNoChange {
+		return err
+	}
+
+	return sourceInstance.Close()
+}

ikuzo/storage/x/postgresql/migrations/000001_create_namespace_table.down.sql

@@ -0,0 +1,7 @@
+BEGIN;
+DROP TABLE IF EXISTS namespace;
+DROP TABLE IF EXISTS namespace_history;
+DROP TRIGGER IF EXISTS namespace_hist_trigger on namespace;
+DROP EXTENSION IF EXISTS temporal_tables;
+DROP EXTENSION IF EXISTS "uuid-ossp";
+END;

ikuzo/storage/x/postgresql/migrations/000001_create_namespace_table.up.sql

@@ -0,0 +1,21 @@
+create extension if not exists temporal_tables;
+
+create extension if not exists "uuid-ossp";
+
+create table namespace (
+    uuid uuid not null default uuid_generate_v1(),
+    prefix text not null,
+    uri text not null,
+    temporary boolean default false,
+    rank integer not null default 1,
+    sys_period tstzrange not null,
+    primary key (prefix, uri)
+);
+
+create table namespace_history (like namespace);
+
+create trigger namespace_hist_trigger 
+    before insert or update or delete 
+        on namespace 
+            for each row
+    execute procedure versioning('sys_period', 'namespace_history', true);

ikuzo/storage/x/postgresql/migrations/000002_add_resource_tables.down.sql

@@ -0,0 +1,11 @@
+drop table if exists resource_audit;
+drop type if exists  operation_t;
+drop trigger if exists resource_audit_trig on resource;
+drop function if exists resource_audit_trig;
+drop table if exists resource;
+drop table if exists dataset;
+drop table if exists organization;
+drop table if exists triple_object;
+drop table if exists datatype;
+drop table if exists predicate;
+drop table if exists metadata_schema;

ikuzo/storage/x/postgresql/migrations/000002_add_resource_tables.up.sql

@@ -0,0 +1,97 @@
+create table metadata_schema (
+    schema_id serial not null primary key,
+    uri text unique not null,
+    json jsonb not null
+);
+
+create table predicate (
+    predicate_id text not null primary key,  -- hashed uri
+    uri text not null unique,
+    search_label text not null,
+    schema_id int references metadata_schema(schema_id)
+);
+
+create table datatype (
+    datatype_id text not null primary key, -- hashed uri
+    uri text not null unique,
+    search_label text not null,
+    comment text not null default 'no comment'
+);
+
+create table triple_object (
+    object_id text not null primary key, -- hashed predicate uri and object (ntriples)
+    object text not null,
+    isResource bool default false,
+    lang varchar(12) not null default '',
+    datatype_id text references datatype(datatype_id),
+    predicate_id text references predicate(predicate_id)
+);
+
+create table organization (
+    org_id text not null primary key,
+    domains text [] not null,
+    rdf_base_url text not null,
+    alt_base_url text []
+);
+
+create table dataset (
+    dataset_id text not null primary key,
+    description text,
+    org_id text references organization(org_id)
+);
+
+create table resource (
+    resource_id text not null primary key, -- hashed uri
+    uri text not null unique,
+    version_id text not null, -- hash of predicate array
+    predicates text [] not null, -- array of triple_object object_ids
+    modified date not null default now(),
+    dataset_id text references dataset(dataset_id)
+);
+
+
+
+create type operation_t as enum ('insert', 'update', 'delete');
+
+create table resource_audit (
+    audit_ts timestamptz not null default now(),
+    operation operation_t not null,
+    dataset_id text references dataset(dataset_id),
+    username text not null default "current_user"(),
+    before jsonb,
+    after jsonb
+);
+
+create or replace function resource_audit_trig()
+ returns trigger
+ language plpgsql
+as $$
+begin
+    if tg_op = 'insert'
+    then
+        insert into resource_audit (operation, after, dataset_id)
+        values (tg_op, to_jsonb(new), new.dataset_id);
+        return new;
+
+    elsif tg_op = 'update'
+    then
+        if new != old then
+            insert into resource_audit (operation, before, after, dataset_id)
+            values (tg_op, to_jsonb(old), to_jsonb(new), new.dataset_id);
+        end if;
+        return new;
+
+    elsif tg_op = 'delete'
+    then
+        insert into audit.users_audit (operation, before, dataset_id)
+        values (tg_op, to_jsonb(old), old.dataset_id);
+        return old;
+    end if;
+end;
+$$;
+
+create trigger resource_audit_trig
+    before insert or update or delete
+        on resource
+            for each row
+ execute procedure resource_audit_trig();