CompanyIQ MCP Server

import pg from 'pg'; import { readFileSync } from 'fs'; import { fileURLToPath } from 'url'; import { dirname, join } from 'path'; import { config } from '../config.js'; const { Pool } = pg; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); export interface CompanyRecord { org_nr: string; name: string; organization_form?: string; nace_code?: string; nace_description?: string; employees_count?: number; established_date?: string; business_address?: string; business_municipality?: string; bankrupt?: boolean; under_liquidation?: boolean; fetched_at?: string; [key: string]: any; } export class CompanyDatabase { private pool: pg.Pool; private initialized: boolean = false; constructor() { this.pool = new Pool({ connectionString: config.database.url, ssl: config.database.ssl ? { rejectUnauthorized: false } : false }); } async init(): Promise<void> { if (this.initialized) return; const schema = readFileSync(join(__dirname, 'schema.sql'), 'utf-8'); await this.pool.query(schema); this.initialized = true; console.log('Database initialized'); } // Expose pool for session store and auth services getPool(): pg.Pool { return this.pool; } // Company operations async insertOrUpdateCompany(company: CompanyRecord): Promise<void> { await this.pool.query(` INSERT INTO companies (org_nr, name, organization_form, organization_form_code, nace_code, nace_description, employees_count, established_date, business_address, business_postcode, business_city, business_municipality, business_municipality_number, postal_address, postal_postcode, postal_city, in_mva_register, in_foretaksregister, in_frivillighetsregister, in_stiftelsesregister, bankrupt, under_liquidation, under_forced_liquidation, last_updated, fetched_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, CURRENT_TIMESTAMP) ON CONFLICT (org_nr) DO UPDATE SET name = EXCLUDED.name, organization_form = EXCLUDED.organization_form, organization_form_code = EXCLUDED.organization_form_code, nace_code = EXCLUDED.nace_code, nace_description = EXCLUDED.nace_description, employees_count = EXCLUDED.employees_count, established_date = EXCLUDED.established_date, business_address = EXCLUDED.business_address, business_postcode = EXCLUDED.business_postcode, business_city = EXCLUDED.business_city, business_municipality = EXCLUDED.business_municipality, business_municipality_number = EXCLUDED.business_municipality_number, postal_address = EXCLUDED.postal_address, postal_postcode = EXCLUDED.postal_postcode, postal_city = EXCLUDED.postal_city, in_mva_register = EXCLUDED.in_mva_register, in_foretaksregister = EXCLUDED.in_foretaksregister, in_frivillighetsregister = EXCLUDED.in_frivillighetsregister, in_stiftelsesregister = EXCLUDED.in_stiftelsesregister, bankrupt = EXCLUDED.bankrupt, under_liquidation = EXCLUDED.under_liquidation, under_forced_liquidation = EXCLUDED.under_forced_liquidation, last_updated = EXCLUDED.last_updated, fetched_at = CURRENT_TIMESTAMP `, [ company.org_nr, company.name, company.organization_form, company.organization_form_code, company.nace_code, company.nace_description, company.employees_count, company.established_date, company.business_address, company.business_postcode, company.business_city, company.business_municipality, company.business_municipality_number, company.postal_address, company.postal_postcode, company.postal_city, company.in_mva_register ? true : false, company.in_foretaksregister ? true : false, company.in_frivillighetsregister ? true : false, company.in_stiftelsesregister ? true : false, company.bankrupt ? true : false, company.under_liquidation ? true : false, company.under_forced_liquidation ? true : false, company.last_updated ]); } async getCompany(orgNr: string): Promise<CompanyRecord | undefined> { const result = await this.pool.query('SELECT * FROM companies WHERE org_nr = $1', [orgNr]); return result.rows[0] as CompanyRecord | undefined; } async searchCompanies(filters: any): Promise<CompanyRecord[]> { const needsCEOFilter = filters.min_ceo_age || filters.max_ceo_age; let query = needsCEOFilter ? 'SELECT DISTINCT c.* FROM companies c INNER JOIN board_members b ON c.org_nr = b.company_org_nr WHERE b.role_type = \'DAGL\' AND b.is_active = true' : 'SELECT * FROM companies WHERE 1=1'; const params: any[] = []; let paramIndex = 1; if (filters.name) { query += needsCEOFilter ? ` AND c.name ILIKE $${paramIndex}` : ` AND name ILIKE $${paramIndex}`; params.push(`%${filters.name}%`); paramIndex++; } if (filters.nace_code) { query += needsCEOFilter ? ` AND c.nace_code LIKE $${paramIndex}` : ` AND nace_code LIKE $${paramIndex}`; params.push(`${filters.nace_code}%`); paramIndex++; } if (filters.municipality) { query += needsCEOFilter ? ` AND c.business_municipality ILIKE $${paramIndex}` : ` AND business_municipality ILIKE $${paramIndex}`; params.push(`%${filters.municipality}%`); paramIndex++; } if (filters.min_employees) { query += needsCEOFilter ? ` AND c.employees_count >= $${paramIndex}` : ` AND employees_count >= $${paramIndex}`; params.push(filters.min_employees); paramIndex++; } if (filters.max_employees) { query += needsCEOFilter ? ` AND c.employees_count <= $${paramIndex}` : ` AND employees_count <= $${paramIndex}`; params.push(filters.max_employees); paramIndex++; } if (filters.min_established_year) { const prefix = needsCEOFilter ? 'c.' : ''; query += ` AND (CAST(SUBSTRING(${prefix}established_date FROM 1 FOR 4) AS INTEGER) >= $${paramIndex} OR CAST(${prefix}established_date AS INTEGER) >= $${paramIndex + 1})`; params.push(filters.min_established_year, filters.min_established_year); paramIndex += 2; } if (filters.max_established_year) { const prefix = needsCEOFilter ? 'c.' : ''; query += ` AND (CAST(SUBSTRING(${prefix}established_date FROM 1 FOR 4) AS INTEGER) <= $${paramIndex} OR CAST(${prefix}established_date AS INTEGER) <= $${paramIndex + 1})`; params.push(filters.max_established_year, filters.max_established_year); paramIndex += 2; } // CEO age filtering if (filters.min_ceo_age) { const currentYear = new Date().getFullYear(); query += ` AND ( CASE WHEN LENGTH(b.birth_date) >= 8 THEN ${currentYear} - CAST(SUBSTRING(b.birth_date FROM 5 FOR 4) AS INTEGER) WHEN LENGTH(b.birth_date) = 6 THEN CASE WHEN CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER) <= ${currentYear % 100} THEN ${currentYear} - (2000 + CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER)) ELSE ${currentYear} - (1900 + CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER)) END ELSE NULL END ) >= $${paramIndex}`; params.push(filters.min_ceo_age); paramIndex++; } if (filters.max_ceo_age) { const currentYear = new Date().getFullYear(); query += ` AND ( CASE WHEN LENGTH(b.birth_date) >= 8 THEN ${currentYear} - CAST(SUBSTRING(b.birth_date FROM 5 FOR 4) AS INTEGER) WHEN LENGTH(b.birth_date) = 6 THEN CASE WHEN CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER) <= ${currentYear % 100} THEN ${currentYear} - (2000 + CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER)) ELSE ${currentYear} - (1900 + CAST(SUBSTRING(b.birth_date FROM 5 FOR 2) AS INTEGER)) END ELSE NULL END ) <= $${paramIndex}`; params.push(filters.max_ceo_age); paramIndex++; } if (filters.exclude_bankrupt) { query += needsCEOFilter ? ' AND (c.bankrupt IS NULL OR c.bankrupt = false)' : ' AND (bankrupt IS NULL OR bankrupt = false)'; } if (filters.in_foretaksregister) { query += needsCEOFilter ? ' AND c.in_foretaksregister = true' : ' AND in_foretaksregister = true'; } query += needsCEOFilter ? ` ORDER BY c.employees_count DESC NULLS LAST LIMIT $${paramIndex}` : ` ORDER BY employees_count DESC NULLS LAST LIMIT $${paramIndex}`; params.push(filters.limit || 100); const result = await this.pool.query(query, params); return result.rows as CompanyRecord[]; } // Board member operations async insertBoardMember(member: any): Promise<void> { await this.pool.query(` INSERT INTO board_members (company_org_nr, person_id, name, role_type, role_description, appointed_date, birth_date, is_active) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT DO NOTHING `, [ member.company_org_nr, member.person_id, member.name, member.role_type, member.role_description, member.appointed_date, member.birth_date, member.is_active ? true : false ]); } async getBoardMembers(orgNr: string): Promise<any[]> { const result = await this.pool.query(` SELECT * FROM board_members WHERE company_org_nr = $1 AND is_active = true ORDER BY CASE role_type WHEN 'DAGL' THEN 1 WHEN 'LEDE' THEN 2 WHEN 'NEST' THEN 3 WHEN 'MEDL' THEN 4 ELSE 5 END `, [orgNr]); return result.rows; } async getCEO(orgNr: string): Promise<any> { const result = await this.pool.query(` SELECT * FROM board_members WHERE company_org_nr = $1 AND role_type = 'DAGL' AND is_active = true LIMIT 1 `, [orgNr]); return result.rows[0]; } calculateAge(birthDate: string): number | null { if (!birthDate) return null; const currentYear = new Date().getFullYear(); const currentYearTwoDigit = currentYear % 100; if (birthDate.length >= 8) { const year = parseInt(birthDate.substring(4, 8)); return currentYear - year; } else if (birthDate.length === 6) { const yearTwoDigit = parseInt(birthDate.substring(4, 6)); const year = yearTwoDigit <= currentYearTwoDigit ? 2000 + yearTwoDigit : 1900 + yearTwoDigit; return currentYear - year; } return null; } // Ownership operations async insertOwnership(ownership: any): Promise<void> { await this.pool.query(` INSERT INTO ownership (company_org_nr, owner_org_nr, owner_name, owner_type, ownership_percentage, shares_count) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING `, [ ownership.company_org_nr, ownership.owner_org_nr, ownership.owner_name, ownership.owner_type, ownership.ownership_percentage, ownership.shares_count ]); } async getOwnership(orgNr: string): Promise<any[]> { const result = await this.pool.query(` SELECT * FROM ownership WHERE company_org_nr = $1 ORDER BY ownership_percentage DESC NULLS LAST `, [orgNr]); return result.rows; } // Subunits operations async insertSubunit(subunit: any): Promise<void> { await this.pool.query(` INSERT INTO subunits (org_nr, parent_org_nr, name, nace_code, employees_count, business_address, municipality, established_date) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (org_nr) DO UPDATE SET parent_org_nr = EXCLUDED.parent_org_nr, name = EXCLUDED.name, nace_code = EXCLUDED.nace_code, employees_count = EXCLUDED.employees_count, business_address = EXCLUDED.business_address, municipality = EXCLUDED.municipality, established_date = EXCLUDED.established_date, fetched_at = CURRENT_TIMESTAMP `, [ subunit.org_nr, subunit.parent_org_nr, subunit.name, subunit.nace_code, subunit.employees_count, subunit.business_address, subunit.municipality, subunit.established_date ]); } async getSubunits(parentOrgNr: string): Promise<any[]> { const result = await this.pool.query(` SELECT * FROM subunits WHERE parent_org_nr = $1 ORDER BY employees_count DESC NULLS LAST `, [parentOrgNr]); return result.rows; } // Financial snapshots async insertFinancialSnapshot(snapshot: any): Promise<void> { await this.pool.query(` INSERT INTO financial_snapshots (org_nr, year, revenue, profit, assets, equity, employees, source) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (org_nr, year) DO UPDATE SET revenue = EXCLUDED.revenue, profit = EXCLUDED.profit, assets = EXCLUDED.assets, equity = EXCLUDED.equity, employees = EXCLUDED.employees, source = EXCLUDED.source, fetched_at = CURRENT_TIMESTAMP `, [ snapshot.org_nr, snapshot.year, snapshot.revenue, snapshot.profit, snapshot.assets, snapshot.equity, snapshot.employees, snapshot.source ]); } async getFinancialHistory(orgNr: string, years: number = 5): Promise<any[]> { const result = await this.pool.query(` SELECT * FROM financial_snapshots WHERE org_nr = $1 ORDER BY year DESC LIMIT $2 `, [orgNr, years]); return result.rows; } // Growth analysis async getGrowthCompanies(naceCode?: string, region?: string, minGrowthPercent: number = 20, limit: number = 50): Promise<any[]> { let query = ` WITH latest_data AS ( SELECT f1.org_nr, f1.year as latest_year, f1.revenue as latest_revenue, f1.employees as latest_employees, f2.revenue as previous_revenue, f2.employees as previous_employees FROM financial_snapshots f1 LEFT JOIN financial_snapshots f2 ON f1.org_nr = f2.org_nr AND f2.year = f1.year - 1 WHERE f1.year = (SELECT MAX(year) FROM financial_snapshots WHERE org_nr = f1.org_nr) ) SELECT c.*, l.latest_revenue, l.latest_employees, l.previous_revenue, l.previous_employees, CAST(((l.latest_revenue - l.previous_revenue) * 100.0 / l.previous_revenue) AS INTEGER) as revenue_growth_percent, CAST(((l.latest_employees - l.previous_employees) * 100.0 / NULLIF(l.previous_employees, 0)) AS INTEGER) as employee_growth_percent FROM companies c JOIN latest_data l ON c.org_nr = l.org_nr WHERE l.previous_revenue > 0 AND ((l.latest_revenue - l.previous_revenue) * 100.0 / l.previous_revenue) >= $1 AND c.bankrupt = false `; const params: any[] = [minGrowthPercent]; let paramIndex = 2; if (naceCode) { query += ` AND c.nace_code LIKE $${paramIndex}`; params.push(`${naceCode}%`); paramIndex++; } if (region) { query += ` AND c.business_municipality ILIKE $${paramIndex}`; params.push(`%${region}%`); paramIndex++; } query += ` ORDER BY revenue_growth_percent DESC LIMIT $${paramIndex}`; params.push(limit); const result = await this.pool.query(query, params); return result.rows; } // Risk scoring async insertRiskScore(score: any): Promise<void> { await this.pool.query(` INSERT INTO risk_scores (org_nr, overall_score, bankruptcy_risk, growth_score, stability_score, financial_health_score, factors, calculated_at) VALUES ($1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP) ON CONFLICT (org_nr) DO UPDATE SET overall_score = EXCLUDED.overall_score, bankruptcy_risk = EXCLUDED.bankruptcy_risk, growth_score = EXCLUDED.growth_score, stability_score = EXCLUDED.stability_score, financial_health_score = EXCLUDED.financial_health_score, factors = EXCLUDED.factors, calculated_at = CURRENT_TIMESTAMP `, [ score.org_nr, score.overall_score, score.bankruptcy_risk, score.growth_score, score.stability_score, score.financial_health_score, JSON.stringify(score.factors) ]); } async getRiskScore(orgNr: string): Promise<any> { const result = await this.pool.query(` SELECT * FROM risk_scores WHERE org_nr = $1 `, [orgNr]); const row = result.rows[0]; if (row && row.factors && typeof row.factors === 'string') { row.factors = JSON.parse(row.factors); } return row; } // Market analysis cache async cacheMarketAnalysis(analysis: any): Promise<void> { await this.pool.query(` INSERT INTO market_analysis (nace_code, analysis_type, region, data, valid_until) VALUES ($1, $2, $3, $4, CURRENT_TIMESTAMP + INTERVAL '${config.cache.ttlHours} hours') `, [ analysis.nace_code, analysis.analysis_type, analysis.region || '', JSON.stringify(analysis.data) ]); } async getMarketAnalysis(naceCode: string, analysisType: string, region?: string): Promise<any> { const result = await this.pool.query(` SELECT * FROM market_analysis WHERE nace_code = $1 AND analysis_type = $2 AND (region = $3 OR $4::text IS NULL) AND valid_until > CURRENT_TIMESTAMP ORDER BY created_at DESC LIMIT 1 `, [naceCode, analysisType, region || '', region || null]); const row = result.rows[0]; if (row && row.data && typeof row.data === 'string') { row.data = JSON.parse(row.data); } return row; } // SSB cache operations async cacheSSBData(cacheEntry: { table_id: string; table_name?: string; category?: string; filters?: any; nace_code?: string; region?: string; year?: string; data: any; metadata?: any; time_series?: any; trend_analysis?: any; }): Promise<void> { const filtersHash = cacheEntry.filters ? JSON.stringify(cacheEntry.filters).split('').reduce((a, b) => { a = ((a << 5) - a) + b.charCodeAt(0); return a & a; }, 0).toString() : 'none'; await this.pool.query(` INSERT INTO ssb_cache (table_id, table_name, category, filters_hash, nace_code, region, year, data, metadata, time_series, trend_analysis, valid_until) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, CURRENT_TIMESTAMP + INTERVAL '${config.cache.ttlHours} hours') ON CONFLICT (table_id, filters_hash) DO UPDATE SET table_name = EXCLUDED.table_name, category = EXCLUDED.category, nace_code = EXCLUDED.nace_code, region = EXCLUDED.region, year = EXCLUDED.year, data = EXCLUDED.data, metadata = EXCLUDED.metadata, time_series = EXCLUDED.time_series, trend_analysis = EXCLUDED.trend_analysis, fetched_at = CURRENT_TIMESTAMP, valid_until = CURRENT_TIMESTAMP + INTERVAL '${config.cache.ttlHours} hours' `, [ cacheEntry.table_id, cacheEntry.table_name || null, cacheEntry.category || null, filtersHash, cacheEntry.nace_code || null, cacheEntry.region || null, cacheEntry.year || null, JSON.stringify(cacheEntry.data), cacheEntry.metadata ? JSON.stringify(cacheEntry.metadata) : null, cacheEntry.time_series ? JSON.stringify(cacheEntry.time_series) : null, cacheEntry.trend_analysis ? JSON.stringify(cacheEntry.trend_analysis) : null ]); } async getSSBCache(table_id: string, filters?: any): Promise<any> { const filtersHash = filters ? JSON.stringify(filters).split('').reduce((a: number, b: string) => { a = ((a << 5) - a) + b.charCodeAt(0); return a & a; }, 0).toString() : 'none'; const result = await this.pool.query(` SELECT * FROM ssb_cache WHERE table_id = $1 AND filters_hash = $2 AND valid_until > CURRENT_TIMESTAMP ORDER BY fetched_at DESC LIMIT 1 `, [table_id, filtersHash]); const row = result.rows[0]; if (row) { return { ...row, data: typeof row.data === 'string' ? JSON.parse(row.data) : row.data, metadata: row.metadata ? (typeof row.metadata === 'string' ? JSON.parse(row.metadata) : row.metadata) : null, time_series: row.time_series ? (typeof row.time_series === 'string' ? JSON.parse(row.time_series) : row.time_series) : null, trend_analysis: row.trend_analysis ? (typeof row.trend_analysis === 'string' ? JSON.parse(row.trend_analysis) : row.trend_analysis) : null }; } return null; } async clearExpiredSSBCache(): Promise<void> { await this.pool.query(` DELETE FROM ssb_cache WHERE valid_until <= CURRENT_TIMESTAMP `); } async close(): Promise<void> { await this.pool.end(); } }