← back to Rentv Adintel
src/connectors/ga4.js
355 lines
'use strict';
/**
* GA4 Data API connector — spec §17.
*
* When GOOGLE_SERVICE_ACCOUNT_JSON_BASE64 is empty (the default in this repo),
* all operations run against the local JSON fixtures with is_demo=true.
*
* When real credentials are present the `// TODO real API` branches document
* the @google-analytics/data call shape but throw a clear error rather than
* quietly failing, because the npm package is intentionally not installed.
*
* Usage:
* const ga4 = require('./ga4');
* await ga4.importAll({ dryRun: false });
*
* @module src/connectors/ga4
*/
const path = require('path');
const fs = require('fs');
const crypto = require('crypto');
const { query, tx } = require('../../db');
// ---------------------------------------------------------------------------
// Credentials probe
// ---------------------------------------------------------------------------
/** Returns true when a real service account credential is configured. */
function hasCredentials() {
return Boolean(process.env.GOOGLE_SERVICE_ACCOUNT_JSON_BASE64);
}
/**
* Test the GA4 connection.
* @returns {{ connected: boolean, demo: boolean, error?: string }}
*/
async function connectionTest() {
if (!hasCredentials()) {
return { connected: false, demo: true, error: 'No GOOGLE_SERVICE_ACCOUNT_JSON_BASE64 configured — running in DEMO/fixture mode.' };
}
// TODO real API: decode the base64 credential, instantiate
// @google-analytics/data BetaAnalyticsDataClient, call runReport with
// a single-day trivial request to validate the property and token.
//
// const { BetaAnalyticsDataClient } = require('@google-analytics/data');
// const creds = JSON.parse(Buffer.from(process.env.GOOGLE_SERVICE_ACCOUNT_JSON_BASE64, 'base64').toString('utf8'));
// const client = new BetaAnalyticsDataClient({ credentials: creds });
// await client.runReport({ property: `properties/${process.env.GA4_PROPERTY_ID}`, dateRanges: [{ startDate: 'yesterday', endDate: 'yesterday' }], metrics: [{ name: 'sessions' }] });
throw new Error('GA4 live API not enabled (no service account)');
}
// ---------------------------------------------------------------------------
// Fixture loader helpers
// ---------------------------------------------------------------------------
const FIXTURES_DIR = path.resolve(__dirname, '../../fixtures');
function loadFixture(name) {
const fp = path.join(FIXTURES_DIR, name);
return JSON.parse(fs.readFileSync(fp, 'utf8'));
}
function fixtureChecksum(name) {
const fp = path.join(FIXTURES_DIR, name);
return crypto.createHash('sha256').update(fs.readFileSync(fp)).digest('hex');
}
// ---------------------------------------------------------------------------
// Import run bookkeeping
// ---------------------------------------------------------------------------
async function startRun(kind, sourceFile, checksum, dryRun) {
const res = await query(
`INSERT INTO analytics_import_runs (kind, source_file, checksum, is_demo, status)
VALUES ($1, $2, $3, true, 'RUNNING') RETURNING id`,
[kind, sourceFile, checksum]
);
return res.rows[0].id;
}
async function finishRun(runId, rowCount, error) {
await query(
`UPDATE analytics_import_runs
SET finished_at = now(),
status = $2,
row_count = $3,
error = $4
WHERE id = $1`,
[runId, error ? 'ERROR' : 'SUCCESS', rowCount, error || null]
);
}
// ---------------------------------------------------------------------------
// Individual importers
// ---------------------------------------------------------------------------
/**
* Upsert rows from ga4-daily.json → ga4_daily_metrics.
* PK is metric_date so this is idempotent.
*
* @param {boolean} dryRun - when true, validate only, no DB writes
* @returns {{ inserted: number, skipped: number, rejected: [], runId: string|null }}
*/
async function importDaily(dryRun = false) {
const fixture = loadFixture('ga4-daily.json');
const rows = fixture.rows;
const checksum = fixtureChecksum('ga4-daily.json');
let runId = null;
let inserted = 0;
if (!dryRun) {
runId = await startRun('GA4_DAILY', 'fixtures/ga4-daily.json', checksum, dryRun);
}
try {
for (const row of rows) {
if (!row.metric_date) continue;
if (!dryRun) {
await query(
`INSERT INTO ga4_daily_metrics
(metric_date, sessions, total_users, new_users, engaged_sessions,
engagement_rate, avg_engagement_time, views, event_count, key_events, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,true)
ON CONFLICT (metric_date) DO UPDATE SET
sessions = EXCLUDED.sessions,
total_users = EXCLUDED.total_users,
new_users = EXCLUDED.new_users,
engaged_sessions = EXCLUDED.engaged_sessions,
engagement_rate = EXCLUDED.engagement_rate,
avg_engagement_time = EXCLUDED.avg_engagement_time,
views = EXCLUDED.views,
event_count = EXCLUDED.event_count,
key_events = EXCLUDED.key_events,
is_demo = true`,
[
row.metric_date, row.sessions, row.total_users, row.new_users,
row.engaged_sessions, row.engagement_rate, row.avg_engagement_time,
row.views, row.event_count, row.key_events,
]
);
}
inserted++;
}
if (!dryRun) await finishRun(runId, inserted, null);
return { inserted, skipped: 0, rejected: [], runId, dryRun };
} catch (err) {
if (!dryRun && runId) await finishRun(runId, inserted, err.message);
throw err;
}
}
/**
* Upsert rows from ga4-landing.json → ga4_landing_page_metrics.
* No natural unique key beyond date+page — we clear-then-insert for the
* fixture's sample date so reruns remain idempotent.
*/
async function importLanding(dryRun = false) {
const fixture = loadFixture('ga4-landing.json');
const rows = fixture.rows;
const checksum = fixtureChecksum('ga4-landing.json');
let runId = null;
let inserted = 0;
if (!dryRun) {
runId = await startRun('GA4_LANDING', 'fixtures/ga4-landing.json', checksum, dryRun);
// Delete existing demo rows for this date range to allow clean re-import
const dates = [...new Set(rows.map((r) => r.metric_date))];
for (const d of dates) {
await query('DELETE FROM ga4_landing_page_metrics WHERE metric_date = $1 AND is_demo = true', [d]);
}
}
try {
for (const row of rows) {
if (!row.metric_date || !row.landing_page) continue;
if (!dryRun) {
await query(
`INSERT INTO ga4_landing_page_metrics
(metric_date, landing_page, sessions, users, views,
engaged_sessions, engagement_rate, key_events, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,true)`,
[
row.metric_date, row.landing_page, row.sessions, row.users,
row.views, row.engaged_sessions, row.engagement_rate, row.key_events,
]
);
}
inserted++;
}
if (!dryRun) await finishRun(runId, inserted, null);
return { inserted, skipped: 0, rejected: [], runId, dryRun };
} catch (err) {
if (!dryRun && runId) await finishRun(runId, inserted, err.message);
throw err;
}
}
/**
* Upsert rows from ga4-geo.json → ga4_geo_metrics.
*/
async function importGeo(dryRun = false) {
const fixture = loadFixture('ga4-geo.json');
const rows = fixture.rows;
const checksum = fixtureChecksum('ga4-geo.json');
let runId = null;
let inserted = 0;
if (!dryRun) {
runId = await startRun('GA4_GEO', 'fixtures/ga4-geo.json', checksum, dryRun);
const dates = [...new Set(rows.map((r) => r.metric_date))];
for (const d of dates) {
await query('DELETE FROM ga4_geo_metrics WHERE metric_date = $1 AND is_demo = true', [d]);
}
}
try {
for (const row of rows) {
if (!row.metric_date) continue;
if (!dryRun) {
await query(
`INSERT INTO ga4_geo_metrics
(metric_date, country, region, city, sessions, users, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,true)`,
[row.metric_date, row.country, row.region, row.city, row.sessions, row.users]
);
}
inserted++;
}
if (!dryRun) await finishRun(runId, inserted, null);
return { inserted, skipped: 0, rejected: [], runId, dryRun };
} catch (err) {
if (!dryRun && runId) await finishRun(runId, inserted, err.message);
throw err;
}
}
/**
* Upsert rows from ga4-acquisition.json → ga4_acquisition_metrics.
*/
async function importAcquisition(dryRun = false) {
const fixture = loadFixture('ga4-acquisition.json');
const rows = fixture.rows;
const checksum = fixtureChecksum('ga4-acquisition.json');
let runId = null;
let inserted = 0;
if (!dryRun) {
runId = await startRun('GA4_ACQUISITION', 'fixtures/ga4-acquisition.json', checksum, dryRun);
const dates = [...new Set(rows.map((r) => r.metric_date))];
for (const d of dates) {
await query('DELETE FROM ga4_acquisition_metrics WHERE metric_date = $1 AND is_demo = true', [d]);
}
}
try {
for (const row of rows) {
if (!row.metric_date) continue;
if (!dryRun) {
await query(
`INSERT INTO ga4_acquisition_metrics
(metric_date, channel_group, session_source, session_medium,
session_campaign, sessions, users, key_events, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,true)`,
[
row.metric_date, row.channel_group, row.session_source,
row.session_medium, row.session_campaign,
row.sessions, row.users, row.key_events,
]
);
}
inserted++;
}
if (!dryRun) await finishRun(runId, inserted, null);
return { inserted, skipped: 0, rejected: [], runId, dryRun };
} catch (err) {
if (!dryRun && runId) await finishRun(runId, inserted, err.message);
throw err;
}
}
// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------
/**
* Import all GA4 fixture tables.
*
* TODO real API: when credentials exist, replace each fixture-load branch with
* a @google-analytics/data runReport call using:
*
* Daily overview dimensions: ['date']
* Daily overview metrics: ['sessions','totalUsers','newUsers','engagedSessions',
* 'engagementRate','averageSessionDuration','screenPageViews',
* 'eventCount','keyEvents']
*
* Landing page dimensions: ['landingPage']
* Landing page metrics: ['sessions','totalUsers','screenPageViews',
* 'engagedSessions','engagementRate','keyEvents']
*
* Geo dimensions: ['country','region','city']
* Geo metrics: ['sessions','totalUsers']
*
* Acquisition dimensions: ['sessionDefaultChannelGroup','sessionSource',
* 'sessionMedium','sessionCampaignName']
* Acquisition metrics: ['sessions','totalUsers','keyEvents']
*
* @param {{ dryRun?: boolean }} options
* @returns {Promise<{ daily: object, landing: object, geo: object, acquisition: object, demo: true }>}
*/
async function importAll({ dryRun = false } = {}) {
if (hasCredentials()) {
throw new Error('GA4 live API not enabled (no service account) — install @google-analytics/data and implement the TODO real API branch.');
}
const [daily, landing, geo, acquisition] = await Promise.all([
importDaily(dryRun),
importLanding(dryRun),
importGeo(dryRun),
importAcquisition(dryRun),
]);
return { daily, landing, geo, acquisition, demo: true };
}
/**
* Ensure an analytics_connections row exists for GA4 (idempotent).
* Sets status=DEMO_CONNECTED when no real creds, NOT_CONNECTED otherwise.
*/
async function ensureConnectionRecord() {
const status = hasCredentials() ? 'NOT_CONNECTED' : 'DEMO_CONNECTED';
await query(
`INSERT INTO analytics_connections (kind, status, is_demo)
VALUES ('GA4', $1, true)
ON CONFLICT DO NOTHING`,
[status]
);
}
module.exports = {
connectionTest,
importAll,
importDaily,
importLanding,
importGeo,
importAcquisition,
ensureConnectionRecord,
hasCredentials,
};