Data Layer

Overview

The Oktuple data layer is built on PostgreSQL with Prisma ORM, providing a robust foundation for project management data. The system uses logical replication for real-time synchronization with search engines and implements comprehensive data validation and integrity constraints.

Database Architecture

Primary Database (PostgreSQL)

Purpose: Main data persistence for all project management operations

Key Features:

• ACID compliance for data integrity
• JSON support for flexible metadata
• Full-text search capabilities
• Logical replication for data sync
• Connection pooling for performance

Database Schema Overview

Core Tables

ENode (Project Structure)

CREATE TABLE enode (
  id VARCHAR PRIMARY KEY,
  owner_id VARCHAR,
  title VARCHAR NOT NULL,
  description TEXT,
  domain VARCHAR,
  prefix VARCHAR,
  type VARCHAR NOT NULL,
  parent_id VARCHAR REFERENCES enode(id),
  template_id VARCHAR REFERENCES template(id),
  template_it_id VARCHAR,
  template_ip_id VARCHAR,
  list_type VARCHAR,
  materialized_path VARCHAR,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  start_at TIMESTAMP,
  end_at TIMESTAMP,
  complete_at TIMESTAMP,
  deleted_at TIMESTAMP,
  meta JSONB,
  archived BOOLEAN DEFAULT FALSE,
  index INTEGER DEFAULT 0,
  position FLOAT,
  old_id VARCHAR,
  has_sprint BOOLEAN DEFAULT FALSE,
  has_limited BOOLEAN DEFAULT TRUE,
  inherit_id VARCHAR,
  count_project INTEGER DEFAULT 0,
  pids VARCHAR[]
);

Issue (Work Items)

CREATE TABLE issue (
  id VARCHAR PRIMARY KEY,
  owner_id VARCHAR NOT NULL,
  title VARCHAR NOT NULL,
  hierarchy_path VARCHAR,
  enode_id VARCHAR REFERENCES enode(id),
  parent_id VARCHAR REFERENCES issue(id),
  effort INTEGER,
  type VARCHAR NOT NULL,
  tags JSONB DEFAULT '[]',
  viewer VARCHAR[],
  people JSONB,
  meta JSONB,
  cfs JSONB,
  position FLOAT,
  addons JSONB,
  stats JSONB,
  template_id VARCHAR REFERENCES template(id),
  old_id VARCHAR,
  abbreviation VARCHAR,
  cover VARCHAR,
  pined BOOLEAN DEFAULT FALSE,
  is_closed BOOLEAN DEFAULT FALSE,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  deleted_at TIMESTAMP,
  completed_at TIMESTAMP,
  due_at TIMESTAMP,
  closed_at TIMESTAMP,
  start_at TIMESTAMP
);

People (Team Members)

CREATE TABLE people (
  id VARCHAR PRIMARY KEY,
  name VARCHAR,
  pl INTEGER,
  parent_id VARCHAR,
  mpath VARCHAR,
  type VARCHAR NOT NULL,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  enode_id VARCHAR NOT NULL REFERENCES enode(id),
  config JSONB,
  old_id VARCHAR
);

Template (Workflow Templates)

CREATE TABLE template (
  id VARCHAR PRIMARY KEY,
  owner_id VARCHAR NOT NULL,
  title VARCHAR,
  origin_id VARCHAR,
  framework VARCHAR,
  rules JSONB,
  enode_id VARCHAR REFERENCES enode(id),
  is_default VARCHAR,
  template_type VARCHAR,
  old_id VARCHAR,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  deleted_at TIMESTAMP
);

Addon (Extended Functionality)

CREATE TABLE addon (
  id VARCHAR,
  type VARCHAR,
  owner_id VARCHAR,
  title VARCHAR NOT NULL,
  parent_id INTEGER,
  metadata JSONB,
  mpath VARCHAR,
  active BOOLEAN DEFAULT TRUE,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  enode_id VARCHAR NOT NULL REFERENCES enode(id),
  is_abstract BOOLEAN DEFAULT FALSE,
  old_id VARCHAR,
  deleted_at TIMESTAMP,
  UNIQUE(enode_id, id)
);

Supporting Tables

Attachment (File Management)

CREATE TABLE attachment (
  id VARCHAR PRIMARY KEY,
  issue_id VARCHAR NOT NULL REFERENCES issue(id),
  key VARCHAR NOT NULL,
  owner_id VARCHAR,
  value JSONB,
  old_id VARCHAR,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW(),
  deleted_at TIMESTAMP
);

Links (Dependencies)

CREATE TABLE links (
  id VARCHAR PRIMARY KEY,
  type VARCHAR NOT NULL,
  from VARCHAR NOT NULL REFERENCES issue(id),
  to VARCHAR NOT NULL REFERENCES issue(id),
  resolve BOOLEAN NOT NULL,
  owner_id VARCHAR NOT NULL,
  old_id VARCHAR,
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW()
);

Role (Permission Management)

CREATE TABLE role (
  id VARCHAR PRIMARY KEY,
  name VARCHAR NOT NULL,
  enode_id VARCHAR NOT NULL REFERENCES enode(id),
  hierarchy_path VARCHAR,
  is_default BOOLEAN DEFAULT FALSE,
  parent_role_id VARCHAR REFERENCES role(id)
);

History (Audit Trail)

CREATE TABLE history (
  id VARCHAR PRIMARY KEY,
  issue_id VARCHAR NOT NULL REFERENCES issue(id),
  owner_id VARCHAR NOT NULL,
  patch JSONB,
  type VARCHAR NOT NULL,
  type_event VARCHAR,
  created_at TIMESTAMP DEFAULT NOW()
);

Data Relationships

Hierarchical Structure

ENode Hierarchy

Root Project
├── Collection 1
│   │   ├── Issue  1
│   │   └── Issue 2
└── Collection 2
    └── Issue 4
        └── Sub Issue 5

Data Validation

Prisma Schema Validation

// Example Prisma model with validation
model enode {
  id                String    @id
  owner_id          String?
  title             String
  description       String?
  domain            String?
  prefix            String?
  type              String
  parent_id         String?
  template_id       String?
  meta              Json?
  archived          Boolean   @default(false)
  created_at        DateTime  @default(now())
  updated_at        DateTime  @updatedAt

  // Relationships
  template          template? @relation(fields: [template_id], references: [id])
  addons            addon[]   @relation("EnodeToAddon")
  issues            issue[]
  peoples           people[]

  @@index([parent_id])
  @@index([template_id])
  @@index([type])
}

Zod Schema Validation

// Request validation schemas
const createEnodeSchema = z.object({
  title: z.string().min(1, "Title is required").max(255),
  description: z.string().optional(),
  type: z.enum(["project", "sprint", "epic", "story", "task"]),
  parent_id: z.string().uuid().optional(),
  template_id: z.string().uuid().optional(),
  domain: z.string().optional(),
  prefix: z.string().max(10).optional(),
  meta: z.record(z.any()).optional(),
});

const updateEnodeSchema = createEnodeSchema.partial();

Data Synchronization

Logical Replication Setup

PostgreSQL Configuration

-- Enable logical replication
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;

-- Create publication
CREATE PUBLICATION oktuple_pub FOR ALL TABLES;

-- Create replication slot
SELECT pg_create_logical_replication_slot('oktuple_syncer_slot', 'pgoutput');

Replication Slot Management

-- Monitor replication slots
SELECT slot_name, active, restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots;

-- Drop replication slot if needed
SELECT pg_drop_replication_slot('oktuple_syncer_slot');

Sync Process Flow

1. Database Change (INSERT/UPDATE/DELETE)
   ↓
2. WAL (Write-Ahead Log) Generation
   ↓
3. Logical Replication Capture
   ↓
4. Syncer Service Processing
   ↓
5. Data Transformation
   ↓
6. Multi-Engine Sync
   ├── Elasticsearch Indexing
   ├── Typesense Indexing
   └── TimescaleDB Indexing
   ↓
7. Fallback Queue (Redis)

Performance Optimization

Database Indexing

Primary Indexes

-- Primary key indexes (automatic)
CREATE INDEX idx_enode_id ON enode(id);
CREATE INDEX idx_issue_id ON issue(id);
CREATE INDEX idx_people_id ON people(id);

Performance Indexes

-- Hierarchical queries
CREATE INDEX idx_enode_parent_id ON enode(parent_id);
CREATE INDEX idx_issue_parent_id ON issue(parent_id);
CREATE INDEX idx_issue_enode_id ON issue(enode_id);

-- Search and filtering
CREATE INDEX idx_enode_type ON enode(type);
CREATE INDEX idx_issue_type ON issue(type);
CREATE INDEX idx_issue_status ON issue(is_closed);

-- Temporal queries
CREATE INDEX idx_enode_created_at ON enode(created_at);
CREATE INDEX idx_issue_created_at ON issue(created_at);
CREATE INDEX idx_issue_due_at ON issue(due_at);

-- JSON field indexing
CREATE INDEX idx_enode_meta ON enode USING GIN (meta);
CREATE INDEX idx_issue_meta ON issue USING GIN (meta);
CREATE INDEX idx_issue_tags ON issue USING GIN (tags);

Composite Indexes

-- Multi-field queries
CREATE INDEX idx_enode_type_parent ON enode(type, parent_id);
CREATE INDEX idx_issue_enode_type ON issue(enode_id, type);
CREATE INDEX idx_people_enode_type ON people(enode_id, type);

Query Optimization

Efficient Hierarchical Queries

-- Using materialized paths
SELECT * FROM enode
WHERE materialized_path LIKE '1.2.%'
ORDER BY materialized_path;

-- Using recursive CTEs
WITH RECURSIVE enode_tree AS (
  SELECT id, title, parent_id, 1 as level
  FROM enode WHERE id = 'root_id'

  UNION ALL

  SELECT e.id, e.title, e.parent_id, et.level + 1
  FROM enode e
  JOIN enode_tree et ON e.parent_id = et.id
)
SELECT * FROM enode_tree ORDER BY level, title;

JSON Field Queries

-- Query JSON metadata
SELECT * FROM enode
WHERE meta->>'priority' = 'high';

-- Array field queries
SELECT * FROM issue
WHERE 'bug' = ANY(tags);

-- Complex JSON queries
SELECT * FROM enode
WHERE meta->>'status' = 'active'
  AND meta->>'priority' IN ('high', 'critical');

Data Integrity

Constraints and Validation

Database Constraints

-- Not null constraints
ALTER TABLE enode ALTER COLUMN title SET NOT NULL;
ALTER TABLE enode ALTER COLUMN type SET NOT NULL;
ALTER TABLE issue ALTER COLUMN title SET NOT NULL;
ALTER TABLE issue ALTER COLUMN owner_id SET NOT NULL;

-- Check constraints
ALTER TABLE issue ADD CONSTRAINT chk_effort_positive
  CHECK (effort IS NULL OR effort > 0);

ALTER TABLE enode ADD CONSTRAINT chk_dates_valid
  CHECK (start_at IS NULL OR end_at IS NULL OR start_at <= end_at);

-- Unique constraints
ALTER TABLE people ADD CONSTRAINT uk_people_enode_user
  UNIQUE (enode_id, id);

Application-Level Validation

// Business rule validation
export class EnodeService {
  async validateEnodeHierarchy(
    parentId: string,
    type: string
  ): Promise<boolean> {
    if (!parentId) return true;

    const parent = await this.prisma.enode.findUnique({
      where: { id: parentId },
    });

    // Validate hierarchy rules
    if (type === "sprint" && parent.type !== "project") {
      throw new Error("Sprints can only be created under projects");
    }

    if (type === "epic" && !["project", "sprint"].includes(parent.type)) {
      throw new Error("Epics can only be created under projects or sprints");
    }

    return true;
  }
}

Transaction Management

ACID Operations

// Example transaction
export class IssueService {
  async createIssueWithDependencies(data: CreateIssueData): Promise<Issue> {
    return await this.prisma.$transaction(async (tx) => {
      // Create issue
      const issue = await tx.issue.create({
        data: {
          title: data.title,
          type: data.type,
          enode_id: data.enode_id,
          owner_id: data.owner_id,
        },
      });

      // Create dependencies
      if (data.dependencies?.length) {
        await tx.links.createMany({
          data: data.dependencies.map((dep) => ({
            type: dep.type,
            from: issue.id,
            to: dep.issue_id,
            resolve: false,
            owner_id: data.owner_id,
          })),
        });
      }

      // Update parent issue stats
      if (data.parent_id) {
        await tx.issue.update({
          where: { id: data.parent_id },
          data: {
            stats: {
              increment: { children_count: 1 },
            },
          },
        });
      }

      return issue;
    });
  }
}

Data Migration

Schema Evolution

Migration Strategy

// Prisma migration example
export async function migrateEnodeStructure() {
  await prisma.$executeRaw`
    -- Add new columns
    ALTER TABLE enode ADD COLUMN IF NOT EXISTS has_sprint BOOLEAN DEFAULT FALSE;
    ALTER TABLE enode ADD COLUMN IF NOT EXISTS has_limited BOOLEAN DEFAULT TRUE;
    
    -- Update existing data
    UPDATE enode SET has_sprint = TRUE WHERE type IN ('project', 'sprint');
    UPDATE enode SET has_limited = FALSE WHERE type = 'project';
    
    -- Add indexes
    CREATE INDEX IF NOT EXISTS idx_enode_has_sprint ON enode(has_sprint);
    CREATE INDEX IF NOT EXISTS idx_enode_has_limited ON enode(has_limited);
  `;
}

Data Transformation

// Data cleanup and transformation
export async function cleanupOrphanedData() {
  // Remove orphaned issues
  await prisma.issue.deleteMany({
    where: {
      enode_id: null,
      deleted_at: { not: null },
    },
  });

  // Remove orphaned attachments
  await prisma.attachment.deleteMany({
    where: {
      issue_id: null,
    },
  });

  // Update materialized paths
  await updateMaterializedPaths();
}

Monitoring and Maintenance

Database Health Checks

Connection Monitoring

export class DatabaseHealthService {
  async checkHealth(): Promise<HealthStatus> {
    try {
      // Test connection
      await this.prisma.$queryRaw`SELECT 1`;

      // Check replication slots
      const slots = await this.prisma.$queryRaw`
        SELECT slot_name, active, restart_lsn 
        FROM pg_replication_slots
      `;

      // Check table sizes
      const sizes = await this.prisma.$queryRaw`
        SELECT schemaname, tablename, pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size
        FROM pg_tables 
        WHERE schemaname = 'public'
        ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
      `;

      return {
        status: "healthy",
        database: "connected",
        replication: slots.length > 0 ? "active" : "inactive",
        tables: sizes.length,
      };
    } catch (error) {
      return {
        status: "unhealthy",
        error: error.message,
      };
    }
  }
}

Performance Monitoring

-- Slow query analysis
SELECT query, calls, total_time, mean_time, rows
FROM pg_stat_statements
ORDER BY mean_time DESC
LIMIT 10;

-- Table access statistics
SELECT schemaname, tablename, seq_scan, seq_tup_read, idx_scan, idx_tup_fetch
FROM pg_stat_user_tables
ORDER BY seq_scan DESC;

-- Index usage statistics
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes
ORDER BY idx_scan DESC;

Backup and Recovery

Backup Strategy

#!/bin/bash
# Automated backup script

# Database backup
pg_dump -h localhost -U postgres -d oktuple > backup_$(date +%Y%m%d_%H%M%S).sql

# Schema backup
pg_dump -h localhost -U postgres -d oktuple --schema-only > schema_$(date +%Y%m%d_%H%M%S).sql

# Data backup (excluding large tables)
pg_dump -h localhost -U postgres -d oktuple \
  --exclude-table=attachment \
  --exclude-table=history \
  > data_$(date +%Y%m%d_%H%M%S).sql

Recovery Procedures

-- Point-in-time recovery
RESTORE DATABASE oktuple FROM 'backup_file'
WITH RECOVERY, STOPAT = '2025-01-01 12:00:00';

-- Restore specific tables
pg_restore -h localhost -U postgres -d oktuple -t enode backup_file.dump
pg_restore -h localhost -U postgres -d oktuple -t issue backup_file.dump

Future Considerations

Planned Enhancements

Database Features

• TimescaleDB: Time-series data for metrics and logs
• Partitioning: Table partitioning for large datasets
• Sharding: Horizontal scaling across multiple databases
• Read Replicas: Load distribution for read operations

Performance Improvements

• Connection Pooling: Advanced connection management
• Query Caching: Result set caching
• Materialized Views: Pre-computed aggregations
• Parallel Queries: Concurrent query execution

Data Management

• Data Archiving: Automatic data lifecycle management
• Compression: Data compression for storage optimization
• Encryption: Field-level encryption for sensitive data
• Audit Logging: Enhanced change tracking