Data Synchronisation Setup

ARCHITECTURE A: HUB-AND-SPOKE (Centralised)

                         +------------------+
                         |                  |
                         |  Central Server  |
                         |  (sync hub)      |
                         |                  |
                         +--------+---------+
                                  |
            +---------------------+---------------------+
            |                     |                     |
    +-------v-------+     +-------v-------+     +-------v-------+
    |               |     |               |     |               |
    |  Field Site A |     |  Field Site B |     |  Field Site C |
    |  (spoke)      |     |  (spoke)      |     |  (spoke)      |
    |               |     |               |     |               |
    +---------------+     +---------------+     +---------------+

- All sync traffic routes through central server
- Field sites never sync directly with each other
- Central server holds authoritative copy
- Simplest conflict resolution (server wins or manual)


ARCHITECTURE B: MULTI-MASTER (Distributed)

    +---------------+           +---------------+
    |               |           |               |
    |  Region Hub 1 |<--------->|  Region Hub 2 |
    |  (master)     |           |  (master)     |
    |               |           |               |
    +-------+-------+           +-------+-------+
            |                           |
    +-------v-------+           +-------v-------+
    |               |           |               |
    |  Field Sites  |           |  Field Sites  |
    |  (replicas)   |           |  (replicas)   |
    |               |           |               |
    +---------------+           +---------------+

- Multiple authoritative nodes sync with each other
- Regional hubs reduce latency for field sites
- Requires vector clocks or similar for ordering
- Complex conflict resolution required


ARCHITECTURE C: EVENT-SOURCED (Append-only)

    +---------------+     +---------------+     +---------------+
    |               |     |               |     |               |
    |  Field Site   |     |  Field Site   |     |  Field Site   |
    |  (events)     |     |  (events)     |     |  (events)     |
    |               |     |               |     |               |
    +-------+-------+     +-------+-------+     +-------+-------+
            |                     |                     |
            +---------------------+---------------------+
                                  |
                         +--------v---------+
                         |                  |
                         |  Event Store     |
                         |  (append-only)   |
                         |                  |
                         +--------+---------+
                                  |
                         +--------v---------+
                         |                  |
                         |  Materialised    |
                         |  Views           |
                         |                  |
                         +------------------+

- Changes recorded as immutable events
- No overwrites, only new events
- Conflicts resolved by event ordering rules
- Full audit trail inherent in design

   -- Create sync tracking schema
   CREATE SCHEMA IF NOT EXISTS sync_control;

   -- Track sync state per client
   CREATE TABLE sync_control.client_state (
       client_id VARCHAR(64) PRIMARY KEY,
       client_name VARCHAR(255) NOT NULL,
       last_sync_timestamp TIMESTAMP WITH TIME ZONE,
       last_sync_version BIGINT DEFAULT 0,
       sync_status VARCHAR(20) DEFAULT 'pending',
       created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
       updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
   );

   -- Track individual table sync progress
   CREATE TABLE sync_control.table_sync_state (
       client_id VARCHAR(64) NOT NULL,
       table_name VARCHAR(128) NOT NULL,
       last_synced_version BIGINT DEFAULT 0,
       last_synced_at TIMESTAMP WITH TIME ZONE,
       row_count_synced BIGINT DEFAULT 0,
       PRIMARY KEY (client_id, table_name),
       FOREIGN KEY (client_id) REFERENCES sync_control.client_state(client_id)
   );

   -- Log sync operations for troubleshooting
   CREATE TABLE sync_control.sync_log (
       log_id BIGSERIAL PRIMARY KEY,
       client_id VARCHAR(64) NOT NULL,
       operation VARCHAR(50) NOT NULL,
       table_name VARCHAR(128),
       rows_affected BIGINT,
       duration_ms INTEGER,
       status VARCHAR(20),
       error_message TEXT,
       created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
   );

   -- Index for performance
   CREATE INDEX idx_sync_log_client_time
       ON sync_control.sync_log(client_id, created_at DESC);

   SELECT table_name FROM information_schema.tables
   WHERE table_schema = 'sync_control';

   -- Add version tracking to existing tables
   -- Example for a beneficiary registration table

   ALTER TABLE programme.beneficiaries
   ADD COLUMN IF NOT EXISTS sync_version BIGINT DEFAULT 0,
   ADD COLUMN IF NOT EXISTS sync_modified_at TIMESTAMP WITH TIME ZONE
       DEFAULT CURRENT_TIMESTAMP,
   ADD COLUMN IF NOT EXISTS sync_deleted BOOLEAN DEFAULT FALSE,
   ADD COLUMN IF NOT EXISTS sync_origin_client VARCHAR(64);

   -- Create trigger to auto-increment version on changes
   CREATE OR REPLACE FUNCTION sync_control.update_sync_version()
   RETURNS TRIGGER AS $$
   BEGIN
       NEW.sync_version := COALESCE(
           (SELECT MAX(sync_version) FROM programme.beneficiaries), 0
       ) + 1;
       NEW.sync_modified_at := CURRENT_TIMESTAMP;
       RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;

   CREATE TRIGGER trg_beneficiaries_sync_version
   BEFORE INSERT OR UPDATE ON programme.beneficiaries
   FOR EACH ROW EXECUTE FUNCTION sync_control.update_sync_version();

   -- Index for efficient delta queries
   CREATE INDEX idx_beneficiaries_sync_version
       ON programme.beneficiaries(sync_version);

   # sync_api.py - Server-side sync endpoints

   from flask import Flask, request, jsonify
   from datetime import datetime
   import psycopg2

   app = Flask(__name__)

   # Configuration
   BATCH_SIZE = 500  # Rows per sync batch
   MAX_SYNC_DURATION = 300  # 5 minutes max per sync session

   @app.route('/api/sync/changes/<table_name>', methods=['GET'])
   def get_changes(table_name):
       """Return rows changed since client's last sync version."""
       client_id = request.headers.get('X-Client-ID')
       since_version = request.args.get('since_version', 0, type=int)

       # Validate table name against allowlist
       allowed_tables = ['beneficiaries', 'distributions', 'assessments']
       if table_name not in allowed_tables:
           return jsonify({'error': 'Table not syncable'}), 400

       conn = get_db_connection()
       cursor = conn.cursor()

       # Fetch changed rows
       cursor.execute(f"""
           SELECT * FROM programme.{table_name}
           WHERE sync_version > %s
           ORDER BY sync_version ASC
           LIMIT %s
       """, (since_version, BATCH_SIZE))

       rows = cursor.fetchall()
       columns = [desc[0] for desc in cursor.description]

       # Get max version in result set
       max_version = max(row[columns.index('sync_version')]
                        for row in rows) if rows else since_version

       # Log the sync operation
       cursor.execute("""
           INSERT INTO sync_control.sync_log
           (client_id, operation, table_name, rows_affected, status)
           VALUES (%s, 'pull', %s, %s, 'success')
       """, (client_id, table_name, len(rows)))

       conn.commit()

       return jsonify({
           'table': table_name,
           'rows': [dict(zip(columns, row)) for row in rows],
           'max_version': max_version,
           'has_more': len(rows) == BATCH_SIZE
       })

   @app.route('/api/sync/push/<table_name>', methods=['POST'])
   def push_changes(table_name):
       """Receive rows from client and merge into server database."""
       client_id = request.headers.get('X-Client-ID')
       data = request.json

       conn = get_db_connection()
       cursor = conn.cursor()

       merged = 0
       conflicts = []

       for row in data['rows']:
           # Check for conflicts (server version newer than client's base)
           cursor.execute(f"""
               SELECT sync_version FROM programme.{table_name}
               WHERE id = %s
           """, (row['id'],))

           server_row = cursor.fetchone()

           if server_row and server_row[0] > row.get('base_version', 0):
               # Conflict detected - record for resolution
               conflicts.append({
                   'id': row['id'],
                   'server_version': server_row[0],
                   'client_version': row.get('base_version', 0)
               })
               continue

           # No conflict - merge the row
           # Use UPSERT to handle both new and updated rows
           columns = [k for k in row.keys() if k != 'base_version']
           values = [row[k] for k in columns]

           placeholders = ', '.join(['%s'] * len(columns))
           update_clause = ', '.join([f"{c} = EXCLUDED.{c}" for c in columns
                                      if c != 'id'])

           cursor.execute(f"""
               INSERT INTO programme.{table_name} ({', '.join(columns)})
               VALUES ({placeholders})
               ON CONFLICT (id) DO UPDATE SET {update_clause}
           """, values)

           merged += 1

       # Update client sync state
       cursor.execute("""
           UPDATE sync_control.client_state
           SET last_sync_timestamp = %s, sync_status = 'active'
           WHERE client_id = %s
       """, (datetime.utcnow(), client_id))

       conn.commit()

       return jsonify({
           'merged': merged,
           'conflicts': conflicts,
           'status': 'partial' if conflicts else 'success'
       })

   -- Register field sites
   INSERT INTO sync_control.client_state (client_id, client_name)
   VALUES
       ('site-nairobi-01', 'Nairobi Field Office'),
       ('site-kampala-01', 'Kampala Field Office'),
       ('site-juba-01', 'Juba Field Office');

   -- Initialise table sync state for each client
   INSERT INTO sync_control.table_sync_state (client_id, table_name)
   SELECT c.client_id, t.table_name
   FROM sync_control.client_state c
   CROSS JOIN (VALUES
       ('beneficiaries'),
       ('distributions'),
       ('assessments')
   ) AS t(table_name);

   SELECT client_id, client_name, sync_status
   FROM sync_control.client_state;

   -- /opt/field-sync/data/local.db
   -- Initialise local database with sync tracking

   CREATE TABLE IF NOT EXISTS sync_queue (
       queue_id INTEGER PRIMARY KEY AUTOINCREMENT,
       table_name TEXT NOT NULL,
       row_id TEXT NOT NULL,
       operation TEXT NOT NULL,  -- 'insert', 'update', 'delete'
       row_data TEXT NOT NULL,   -- JSON serialised row
       base_version INTEGER,
       queued_at TEXT DEFAULT (datetime('now')),
       sync_attempts INTEGER DEFAULT 0,
       last_attempt TEXT,
       status TEXT DEFAULT 'pending'
   );

   CREATE TABLE IF NOT EXISTS sync_state (
       table_name TEXT PRIMARY KEY,
       last_synced_version INTEGER DEFAULT 0,
       last_synced_at TEXT,
       rows_pending INTEGER DEFAULT 0
   );

   CREATE INDEX idx_queue_status ON sync_queue(status, table_name);

   -- Initialise sync state for each table
   INSERT OR IGNORE INTO sync_state (table_name)
   VALUES ('beneficiaries'), ('distributions'), ('assessments');

   # Add to sync_client.py CONFIG

   BANDWIDTH_CONFIG = {
       'max_upload_kbps': 512,      # Limit upload to 512 Kbps
       'max_download_kbps': 1024,   # Limit download to 1 Mbps
       'chunk_size_kb': 64,         # Transfer in 64KB chunks
       'inter_chunk_delay_ms': 50,  # Delay between chunks
       'peak_hours': [(9, 12), (14, 17)],  # Reduce during peak
       'peak_reduction_pct': 50     # Use 50% of limits during peak
   }

   def get_current_limits():
       """Return bandwidth limits adjusted for time of day."""
       from datetime import datetime
       hour = datetime.now().hour

       in_peak = any(start <= hour < end
                    for start, end in BANDWIDTH_CONFIG['peak_hours'])

       multiplier = (100 - BANDWIDTH_CONFIG['peak_reduction_pct']) / 100 if in_peak else 1.0

       return {
           'upload': int(BANDWIDTH_CONFIG['max_upload_kbps'] * multiplier),
           'download': int(BANDWIDTH_CONFIG['max_download_kbps'] * multiplier)
       }

   -- Add conflict detection configuration table
   CREATE TABLE sync_control.conflict_rules (
       table_name VARCHAR(128) PRIMARY KEY,
       detection_method VARCHAR(50) NOT NULL,
       resolution_strategy VARCHAR(50) NOT NULL,
       priority_field VARCHAR(64),
       merge_fields TEXT[],  -- Fields that can be auto-merged
       notify_on_conflict BOOLEAN DEFAULT TRUE
   );

   -- Configure per-table rules
   INSERT INTO sync_control.conflict_rules VALUES
   -- Beneficiary data: server wins, notify for review
   ('beneficiaries', 'version_compare', 'server_wins', NULL,
    ARRAY['phone_secondary', 'notes'], TRUE),

   -- Distributions: last-write-wins by timestamp
   ('distributions', 'timestamp_compare', 'last_write_wins',
    'sync_modified_at', NULL, FALSE),

   -- Assessments: require manual resolution
   ('assessments', 'version_compare', 'manual', NULL, NULL, TRUE);

   def detect_conflict(table_name, client_row, server_row, cursor):
       """Determine if a conflict exists and how to handle it."""

       # Get conflict rules for this table
       cursor.execute("""
           SELECT detection_method, resolution_strategy,
                  priority_field, merge_fields
           FROM sync_control.conflict_rules
           WHERE table_name = %s
       """, (table_name,))

       rule = cursor.fetchone()
       if not rule:
           # Default: version comparison, server wins
           return detect_version_conflict(client_row, server_row)

       method, strategy, priority_field, merge_fields = rule

       if method == 'version_compare':
           if server_row['sync_version'] > client_row.get('base_version', 0):
               return {'is_conflict': True, 'strategy': strategy}
           return {'is_conflict': False}

       elif method == 'timestamp_compare':
           server_time = server_row[priority_field]
           client_time = client_row.get(priority_field)

           if server_time and client_time and server_time > client_time:
               return {'is_conflict': True, 'strategy': strategy}
           return {'is_conflict': False}

       return {'is_conflict': False}

   def resolve_conflict(strategy, client_row, server_row, merge_fields):
       """Apply resolution strategy and return merged row."""

       if strategy == 'server_wins':
           return server_row

       elif strategy == 'client_wins':
           return client_row

       elif strategy == 'last_write_wins':
           if client_row.get('sync_modified_at', '') > server_row.get('sync_modified_at', ''):
               return client_row
           return server_row

       elif strategy == 'field_merge' and merge_fields:
           # Merge specified fields from client, keep rest from server
           merged = server_row.copy()
           for field in merge_fields:
               if field in client_row and client_row[field]:
                   merged[field] = client_row[field]
           return merged

       elif strategy == 'manual':
           # Record for manual resolution
           return None

       return server_row  # Default fallback

   # Add to sync_api.py

   def notify_conflict(table_name, record_id, client_id, conflict_details):
       """Send notification for manual conflict resolution."""

       notification = {
           'type': 'sync_conflict',
           'table': table_name,
           'record_id': record_id,
           'client': client_id,
           'server_version': conflict_details['server_version'],
           'client_version': conflict_details['client_version'],
           'timestamp': datetime.utcnow().isoformat()
       }

       # Store in notification queue
       conn = get_db_connection()
       cursor = conn.cursor()
       cursor.execute("""
           INSERT INTO sync_control.conflict_notifications
           (table_name, record_id, client_id, details, status)
           VALUES (%s, %s, %s, %s, 'pending')
       """, (table_name, record_id, client_id, json.dumps(notification)))
       conn.commit()

       # Send email to data manager
       send_email(
           to='data-manager@example.org',
           subject=f'Sync conflict requires resolution: {table_name}',
           body=f"""
           A sync conflict occurred that requires manual resolution.

           Table: {table_name}
           Record ID: {record_id}
           Field Client: {client_id}

           Please review in the data management interface.
           """
       )

   -- Add priority configuration
   ALTER TABLE sync_control.table_sync_state
   ADD COLUMN sync_priority INTEGER DEFAULT 50,
   ADD COLUMN sync_direction VARCHAR(20) DEFAULT 'bidirectional',
   ADD COLUMN sync_filter TEXT;

   -- Set priorities (lower number = higher priority)
   UPDATE sync_control.table_sync_state SET sync_priority = 10
   WHERE table_name = 'beneficiaries';  -- Critical registration data

   UPDATE sync_control.table_sync_state SET sync_priority = 20
   WHERE table_name = 'distributions';   -- Distribution records

   UPDATE sync_control.table_sync_state SET sync_priority = 30
   WHERE table_name = 'assessments';     -- Assessment forms

   -- Sync only active beneficiaries for each site's region
   UPDATE sync_control.table_sync_state
   SET sync_filter = 'region_code = ''NAIROBI'' AND status = ''active'''
   WHERE client_id = 'site-nairobi-01' AND table_name = 'beneficiaries';

   -- Sync only recent distributions (last 90 days)
   UPDATE sync_control.table_sync_state
   SET sync_filter = 'distribution_date > CURRENT_DATE - INTERVAL ''90 days'''
   WHERE table_name = 'distributions';

   def get_changes_with_filter(table_name, client_id, since_version):
       cursor = get_db_connection().cursor()

       # Get client-specific filter
       cursor.execute("""
           SELECT sync_filter FROM sync_control.table_sync_state
           WHERE client_id = %s AND table_name = %s
       """, (client_id, table_name))

       row = cursor.fetchone()
       filter_clause = f"AND ({row[0]})" if row and row[0] else ""

       query = f"""
           SELECT * FROM programme.{table_name}
           WHERE sync_version > %s {filter_clause}
           ORDER BY sync_version ASC
           LIMIT %s
       """

       cursor.execute(query, (since_version, BATCH_SIZE))
       return cursor.fetchall()

   -- View: Current sync status across all clients
   CREATE VIEW sync_control.sync_health AS
   SELECT
       cs.client_id,
       cs.client_name,
       cs.last_sync_timestamp,
       EXTRACT(EPOCH FROM (NOW() - cs.last_sync_timestamp))/3600 AS hours_since_sync,
       cs.sync_status,
       COUNT(CASE WHEN sq.status = 'pending' THEN 1 END) AS pending_changes,
       COUNT(CASE WHEN sq.status = 'conflict' THEN 1 END) AS unresolved_conflicts
   FROM sync_control.client_state cs
   LEFT JOIN sync_control.sync_queue sq ON cs.client_id = sq.client_id
   GROUP BY cs.client_id, cs.client_name, cs.last_sync_timestamp, cs.sync_status;

   -- View: Sync performance metrics
   CREATE VIEW sync_control.sync_metrics AS
   SELECT
       client_id,
       DATE(created_at) AS sync_date,
       COUNT(*) AS sync_operations,
       SUM(rows_affected) AS total_rows,
       AVG(duration_ms) AS avg_duration_ms,
       COUNT(CASE WHEN status = 'error' THEN 1 END) AS errors
   FROM sync_control.sync_log
   WHERE created_at > NOW() - INTERVAL '30 days'
   GROUP BY client_id, DATE(created_at);

   -- Function to check sync health and generate alerts
   CREATE OR REPLACE FUNCTION sync_control.check_sync_alerts()
   RETURNS TABLE (
       alert_type VARCHAR,
       client_id VARCHAR,
       message TEXT
   ) AS $$
   BEGIN
       -- Alert: Client not synced in 24 hours
       RETURN QUERY
       SELECT
           'sync_stale'::VARCHAR,
           cs.client_id,
           format('Client %s has not synced in %s hours',
                  cs.client_name,
                  ROUND(EXTRACT(EPOCH FROM (NOW() - cs.last_sync_timestamp))/3600))
       FROM sync_control.client_state cs
       WHERE cs.last_sync_timestamp < NOW() - INTERVAL '24 hours';

       -- Alert: High conflict rate
       RETURN QUERY
       SELECT
           'high_conflicts'::VARCHAR,
           sl.client_id,
           format('Client %s has %s unresolved conflicts', sl.client_id, conflict_count)
       FROM (
           SELECT client_id, COUNT(*) AS conflict_count
           FROM sync_control.sync_log
           WHERE status = 'conflict'
           AND created_at > NOW() - INTERVAL '7 days'
           GROUP BY client_id
           HAVING COUNT(*) > 10
       ) sl;

       -- Alert: Repeated sync failures
       RETURN QUERY
       SELECT
           'sync_errors'::VARCHAR,
           sl.client_id,
           format('Client %s has %s sync errors in last 24 hours', sl.client_id, error_count)
       FROM (
           SELECT client_id, COUNT(*) AS error_count
           FROM sync_control.sync_log
           WHERE status = 'error'
           AND created_at > NOW() - INTERVAL '24 hours'
           GROUP BY client_id
           HAVING COUNT(*) > 5
       ) sl;
   END;
   $$ LANGUAGE plpgsql;

   -- Daily sync health dashboard
   SELECT
       client_name,
       CASE
           WHEN hours_since_sync < 4 THEN 'healthy'
           WHEN hours_since_sync < 24 THEN 'warning'
           ELSE 'critical'
       END AS status,
       hours_since_sync,
       pending_changes,
       unresolved_conflicts
   FROM sync_control.sync_health
   ORDER BY hours_since_sync DESC;

-- Verify sync schema exists
SELECT COUNT(*) FROM information_schema.tables
WHERE table_schema = 'sync_control';
-- Expected: 4 (client_state, table_sync_state, sync_log, conflict_rules)

-- Verify clients registered
SELECT client_id, sync_status FROM sync_control.client_state;
-- Expected: All configured clients listed with 'pending' status

-- Verify version tracking on application tables
SELECT column_name FROM information_schema.columns
WHERE table_name = 'beneficiaries' AND column_name LIKE 'sync_%';
-- Expected: sync_version, sync_modified_at, sync_deleted, sync_origin_client