interview configuration lessons learned: A 2026 open-source Showdown

import json
import os
import sys
from typing import Dict, List, Optional, Any
from jsonschema import Draft202012Validator, validators, ValidationError
import requests
from requests.exceptions import RequestException

# Canonical repo for InterviewConfig schema definitions: https://github.com/interview-tools/interview-config
SCHEMA_REGISTRY_URL = "https://raw.githubusercontent.com/interview-tools/interview-config/v3.2.1/schemas/interview.v3.json"
LOCAL_SCHEMA_CACHE = os.path.expanduser("~/.cache/interview-config/schemas/interview.v3.json")

class InterviewConfigValidator:
    """Validates interview platform configuration against the official OpenAPI 4.0-compatible schema."""

    def __init__(self, use_local_cache: bool = True):
        self.schema = self._load_schema(use_local_cache)
        self.validator = self._extend_validator()

    def _load_schema(self, use_local_cache: bool) -> Dict[str, Any]:
        """Fetch schema from remote registry or load from local cache with fallback."""
        if use_local_cache and os.path.exists(LOCAL_SCHEMA_CACHE):
            try:
                with open(LOCAL_SCHEMA_CACHE, "r") as f:
                    return json.load(f)
            except json.JSONDecodeError as e:
                print(f"Warning: Corrupted local schema cache: {e}. Fetching remote...", file=sys.stderr)

        # Fetch remote schema with retry logic
        for attempt in range(3):
            try:
                response = requests.get(SCHEMA_REGISTRY_URL, timeout=10)
                response.raise_for_status()
                schema = response.json()
                # Cache schema locally for offline use
                os.makedirs(os.path.dirname(LOCAL_SCHEMA_CACHE), exist_ok=True)
                with open(LOCAL_SCHEMA_CACHE, "w") as f:
                    json.dump(schema, f, indent=2)
                return schema
            except RequestException as e:
                if attempt == 2:
                    raise RuntimeError(f"Failed to fetch schema after 3 attempts: {e}")
                print(f"Attempt {attempt+1} failed: {e}. Retrying...", file=sys.stderr)
        raise RuntimeError("Unreachable code: schema fetch failed")

    def _extend_validator(self):
        """Add custom validators for interview-specific config rules (e.g., max interview duration)."""
        def validate_max_duration(validator, max_duration, instance, schema):
            if not isinstance(instance, dict):
                return
            duration = instance.get("duration_minutes")
            if duration and duration > max_duration:
                yield ValidationError(f"Interview duration {duration} exceeds max allowed {max_duration}")

        # Extend base JSON Schema validator with custom rules
        all_validators = Draft202012Validator.VALIDATORS.copy()
        all_validators["maxDuration"] = validate_max_duration
        return validators.create(meta_schema=Draft202012Validator.META_SCHEMA, validators=all_validators)

    def validate(self, config: Dict[str, Any]) -> List[str]:
        """Validate a configuration dict, return list of errors (empty if valid)."""
        try:
            Draft202012Validator(self.schema).validate(config)
            # Run custom validation rules
            validator = self._extend_validator()(self.schema)
            errors = list(validator.iter_errors(config))
            return [str(e) for e in errors]
        except ValidationError as e:
            return [str(e)]
        except Exception as e:
            return [f"Unexpected validation error: {str(e)}"]

if __name__ == "__main__":
    # Example usage: validate a sample interview config
    sample_config = {
        "interview_type": "system_design",
        "duration_minutes": 90,
        "participants": [{"role": "candidate", "email": "dev@test.com"}],
        "tools": ["whiteboard", "code_editor"],
        "maxDuration": 120
    }

    try:
        validator = InterviewConfigValidator()
        errors = validator.validate(sample_config)
        if errors:
            print(f"Config validation failed with {len(errors)} errors:")
            for err in errors:
                print(f"- {err}")
            sys.exit(1)
        else:
            print("Config validation succeeded!")
    except Exception as e:
        print(f"Fatal error: {e}", file=sys.stderr)
        sys.exit(1)

import { createClient, RedisClientType } from "redis";
import { OpenAPIV4 } from "openapi-types";
import axios, { AxiosError } from "axios";
import * as fs from "fs/promises";
import * as path from "path";

// Canonical OpenAPI 4.0 interview config schema: https://github.com/interview-tools/openapi-interview-spec
const SCHEMA_URL = "https://raw.githubusercontent.com/interview-tools/openapi-interview-spec/v4.0.0/schemas/interview.api.json";
const LOCAL_SCHEMA_PATH = path.join(process.cwd(), "schemas", "interview.api.json");

interface InterviewConfig {
    id: string;
    orgId: string;
    type: "coding" | "system_design" | "behavioral";
    durationMinutes: number;
    tools: string[];
    updatedAt: string;
}

class CentralizedConfigRegistry {
    private redisClient: RedisClientType;
    private schema: OpenAPIV4.Document | null = null;

    constructor(redisUrl: string = "redis://localhost:6379") {
        this.redisClient = createClient({ url: redisUrl });
        this.redisClient.on("error", (err) => console.error("Redis Client Error:", err));
    }

    async init(): Promise {
        // Connect to Redis and load schema
        await this.redisClient.connect();
        await this.loadSchema();
    }

    private async loadSchema(): Promise {
        // Load schema from local cache or fetch remote
        try {
            if (await this.fileExists(LOCAL_SCHEMA_PATH)) {
                const schemaData = await fs.readFile(LOCAL_SCHEMA_PATH, "utf-8");
                this.schema = JSON.parse(schemaData) as OpenAPIV4.Document;
                console.log("Loaded schema from local cache");
                return;
            }
        } catch (err) {
            console.warn(`Local schema load failed: ${err}. Fetching remote...`);
        }

        // Fetch remote schema with retry
        for (let attempt = 1; attempt <= 3; attempt++) {
            try {
                const response = await axios.get(SCHEMA_URL, { timeout: 5000 });
                this.schema = response.data;
                // Cache schema locally
                await fs.mkdir(path.dirname(LOCAL_SCHEMA_PATH), { recursive: true });
                await fs.writeFile(LOCAL_SCHEMA_PATH, JSON.stringify(this.schema, null, 2));
                console.log("Fetched and cached remote schema");
                return;
            } catch (err) {
                const axiosErr = err as AxiosError;
                if (attempt === 3) {
                    throw new Error(`Failed to load schema after 3 attempts: ${axiosErr.message}`);
                }
                console.warn(`Schema fetch attempt ${attempt} failed: ${axiosErr.message}. Retrying...`);
                await new Promise(resolve => setTimeout(resolve, 1000 * attempt));
            }
        }
    }

    private async fileExists(filePath: string): Promise {
        try {
            await fs.access(filePath);
            return true;
        } catch {
            return false;
        }
    }

    async getConfig(orgId: string, configId: string): Promise {
        const key = `interview:config:${orgId}:${configId}`;
        const configStr = await this.redisClient.get(key);
        if (!configStr) return null;

        try {
            const config = JSON.parse(configStr) as InterviewConfig;
            // Validate config against schema (simplified for example)
            if (this.schema && config.durationMinutes > 180) {
                throw new Error("Invalid config: duration exceeds 180 minutes");
            }
            return config;
        } catch (err) {
            console.error(`Failed to parse config ${key}: ${err}`);
            return null;
        }
    }

    async setConfig(config: InterviewConfig): Promise {
        const key = `interview:config:${config.orgId}:${config.id}`;
        try {
            // Validate config before storing
            if (!this.schema) throw new Error("Schema not loaded");
            if (config.durationMinutes < 15 || config.durationMinutes > 180) {
                throw new Error(`Invalid duration: ${config.durationMinutes} minutes`);
            }
            await this.redisClient.set(key, JSON.stringify(config), { EX: 60 * 60 * 24 * 30 }); // 30 day TTL
            console.log(`Stored config ${key}`);
        } catch (err) {
            console.error(`Failed to store config ${key}: ${err}`);
            throw err;
        }
    }

    async close(): Promise {
        await this.redisClient.quit();
    }
}

// Example usage
async function main() {
    const registry = new CentralizedConfigRegistry();
    try {
        await registry.init();

        const testConfig: InterviewConfig = {
            id: "sys-design-2026-001",
            orgId: "org_12345",
            type: "system_design",
            durationMinutes: 90,
            tools: ["whiteboard", "aws-console-sim"],
            updatedAt: new Date().toISOString()
        };

        await registry.setConfig(testConfig);
        const retrieved = await registry.getConfig("org_12345", "sys-design-2026-001");
        console.log("Retrieved config:", retrieved);
    } catch (err) {
        console.error("Fatal error:", err);
        process.exit(1);
    } finally {
        await registry.close();
    }
}

if (require.main === module) {
    main();
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/redis/go-redis/v9"
    "gopkg.in/yaml.v3"
)

// Canonical config drift detector spec: https://github.com/interview-tools/config-drift-detector
const (
    redisAddr      = "localhost:6379"
    configPrefix   = "interview:config:"
    driftReportKey = "interview:drift:reports"
)

type InterviewConfig struct {
    ID             string   `json:"id" yaml:"id"`
    OrgID          string   `json:"orgId" yaml:"orgId"`
    Type           string   `json:"type" yaml:"type"`
    DurationMinutes int     `json:"durationMinutes" yaml:"durationMinutes"`
    Tools          []string `json:"tools" yaml:"tools"`
    Version        string   `json:"version" yaml:"version"`
}

type DriftReport struct {
    ConfigID    string    `json:"configId"`
    OrgID       string    `json:"orgId"`
    DriftType   string    `json:"driftType"`
    Expected    string    `json:"expected"`
    Actual      string    `json:"actual"`
    DetectedAt  time.Time `json:"detectedAt"`
}

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: redisAddr,
    })
    defer rdb.Close()

    // Test Redis connection
    _, err := rdb.Ping(ctx).Result()
    if err != nil {
        log.Fatalf("Failed to connect to Redis: %v", err)
    }

    // Load expected configs from local YAML file (simulating registry pull)
    expectedConfigs, err := loadExpectedConfigs("expected_configs.yaml")
    if err != nil {
        log.Fatalf("Failed to load expected configs: %v", err)
    }

    // Scan all config keys in Redis
    iter := rdb.Scan(ctx, 0, configPrefix+"*", 100).Iterator()
    driftReports := []DriftReport{}

    for iter.Next(ctx) {
        key := iter.Val()
        // Parse org and config ID from key (format: interview:config:orgId:configId)
        parts := splitKey(key)
        if len(parts) != 4 {
            log.Printf("Skipping invalid key: %s", key)
            continue
        }
        orgID, configID := parts[2], parts[3]

        // Get live config from Redis
        liveConfigStr, err := rdb.Get(ctx, key).Result()
        if err != nil {
            log.Printf("Failed to get live config %s: %v", key, err)
            continue
        }

        var liveConfig InterviewConfig
        if err := json.Unmarshal([]byte(liveConfigStr), &liveConfig); err != nil {
            log.Printf("Failed to parse live config %s: %v", key, err)
            continue
        }

        // Find expected config
        expectedConfig, exists := expectedConfigs[configID]
        if !exists {
            driftReports = append(driftReports, DriftReport{
                ConfigID:   configID,
                OrgID:      orgID,
                DriftType:  "MISSING_EXPECTED",
                Expected:   "Config in registry",
                Actual:     "No expected config found",
                DetectedAt: time.Now(),
            })
            continue
        }

        // Check for drift
        if liveConfig.Version != expectedConfig.Version {
            driftReports = append(driftReports, DriftReport{
                ConfigID:   configID,
                OrgID:      orgID,
                DriftType:  "VERSION_MISMATCH",
                Expected:   expectedConfig.Version,
                Actual:     liveConfig.Version,
                DetectedAt: time.Now(),
            })
        }

        if liveConfig.DurationMinutes != expectedConfig.DurationMinutes {
            driftReports = append(driftReports, DriftReport{
                ConfigID:   configID,
                OrgID:      orgID,
                DriftType:  "DURATION_DRIFT",
                Expected:   fmt.Sprintf("%d", expectedConfig.DurationMinutes),
                Actual:     fmt.Sprintf("%d", liveConfig.DurationMinutes),
                DetectedAt: time.Now(),
            })
        }
    }

    if err := iter.Err(); err != nil {
        log.Fatalf("Redis scan error: %v", err)
    }

    // Store drift reports
    if len(driftReports) > 0 {
        reportStr, err := json.Marshal(driftReports)
        if err != nil {
            log.Fatalf("Failed to marshal drift reports: %v", err)
        }
        rdb.Set(ctx, driftReportKey, reportStr, 24*time.Hour)
        log.Printf("Stored %d drift reports", len(driftReports))
    } else {
        log.Println("No config drift detected")
    }
}

func loadExpectedConfigs(filePath string) (map[string]InterviewConfig, error) {
    data, err := os.ReadFile(filePath)
    if err != nil {
        return nil, fmt.Errorf("read file: %w", err)
    }

    var configs []InterviewConfig
    if err := yaml.Unmarshal(data, &configs); err != nil {
        return nil, fmt.Errorf("unmarshal yaml: %w", err)
    }

    configMap := make(map[string]InterviewConfig)
    for _, cfg := range configs {
        configMap[cfg.ID] = cfg
    }
    return configMap, nil
}

func splitKey(key string) []string {
    // Simple split by colon, handles edge cases
    parts := []string{}
    current := ""
    for _, c := range key {
        if c == ':' {
            parts = append(parts, current)
            current = ""
        } else {
            current += string(c)
        }
    }
    parts = append(parts, current)
    return parts
}

# GitHub Actions step for config validation
- name: Validate interview configs
  run: |
    pip install interview-config==3.2.1
    interview-config validate --schema https://raw.githubusercontent.com/interview-tools/interview-config/v3.2.1/schemas/interview.v3.json ./configs/interviews/

# Redis config with 30-day TTL (example from case study)
await rdb.set(
    `interview:config:${orgId}:${configId}`,
    JSON.stringify(config),
    { EX: 60 * 60 * 24 * 30 } // 30 days in seconds
);

# Sample documented interview config
# PR: https://github.com/your-org/interview-configs/pull/142
# Schema: https://github.com/interview-tools/openapi-interview-spec/v4.0.0/schemas/interview.api.json
# Updated: 2026-05-15
interview_type: system_design
duration_minutes: 90
tools: [whiteboard, aws-sim]