feat: Backup & Restore - Manual backup tested complete.
This commit is contained in:
Eric Gullickson
@@ -1,436 +0,0 @@
|
||||
/**
|
||||
* @ai-summary Station oversight business logic for admin operations
|
||||
* @ai-context Manages global stations and user-saved stations with audit logging
|
||||
*/
|
||||
|
||||
import { Pool } from 'pg';
|
||||
import { redis } from '../../../core/config/redis';
|
||||
import { logger } from '../../../core/logging/logger';
|
||||
import { AdminRepository } from '../data/admin.repository';
|
||||
import { StationsRepository } from '../../stations/data/stations.repository';
|
||||
import { Station, SavedStation } from '../../stations/domain/stations.types';
|
||||
|
||||
interface CreateStationData {
|
||||
placeId: string;
|
||||
name: string;
|
||||
address: string;
|
||||
latitude: number;
|
||||
longitude: number;
|
||||
priceRegular?: number;
|
||||
pricePremium?: number;
|
||||
priceDiesel?: number;
|
||||
rating?: number;
|
||||
photoReference?: string;
|
||||
}
|
||||
|
||||
interface UpdateStationData {
|
||||
name?: string;
|
||||
address?: string;
|
||||
latitude?: number;
|
||||
longitude?: number;
|
||||
priceRegular?: number;
|
||||
pricePremium?: number;
|
||||
priceDiesel?: number;
|
||||
rating?: number;
|
||||
photoReference?: string;
|
||||
}
|
||||
|
||||
interface StationListResult {
|
||||
total: number;
|
||||
stations: Station[];
|
||||
}
|
||||
|
||||
export class StationOversightService {
|
||||
private stationsRepository: StationsRepository;
|
||||
|
||||
constructor(
|
||||
private pool: Pool,
|
||||
private adminRepository: AdminRepository
|
||||
) {
|
||||
this.stationsRepository = new StationsRepository(pool);
|
||||
}
|
||||
|
||||
/**
|
||||
* List all stations globally with pagination and search
|
||||
*/
|
||||
async listAllStations(
|
||||
limit: number = 100,
|
||||
offset: number = 0,
|
||||
search?: string
|
||||
): Promise<StationListResult> {
|
||||
try {
|
||||
let countQuery = 'SELECT COUNT(*) as total FROM station_cache';
|
||||
let dataQuery = `
|
||||
SELECT
|
||||
id, place_id, name, address, latitude, longitude,
|
||||
price_regular, price_premium, price_diesel, rating, photo_reference, cached_at
|
||||
FROM station_cache
|
||||
`;
|
||||
const params: any[] = [];
|
||||
|
||||
// Add search filter if provided
|
||||
if (search) {
|
||||
const searchCondition = ` WHERE name ILIKE $1 OR address ILIKE $1`;
|
||||
countQuery += searchCondition;
|
||||
dataQuery += searchCondition;
|
||||
params.push(`%${search}%`);
|
||||
}
|
||||
|
||||
dataQuery += ' ORDER BY cached_at DESC LIMIT $' + (params.length + 1) + ' OFFSET $' + (params.length + 2);
|
||||
params.push(limit, offset);
|
||||
|
||||
const [countResult, dataResult] = await Promise.all([
|
||||
this.pool.query(countQuery, search ? [`%${search}%`] : []),
|
||||
this.pool.query(dataQuery, params),
|
||||
]);
|
||||
|
||||
const total = parseInt(countResult.rows[0].total, 10);
|
||||
const stations = dataResult.rows.map(row => this.mapStationRow(row));
|
||||
|
||||
return { total, stations };
|
||||
} catch (error) {
|
||||
logger.error('Error listing all stations', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new station in the cache
|
||||
*/
|
||||
async createStation(
|
||||
actorId: string,
|
||||
data: CreateStationData
|
||||
): Promise<Station> {
|
||||
try {
|
||||
// Create station using repository
|
||||
const station: Station = {
|
||||
id: '', // Will be generated by database
|
||||
placeId: data.placeId,
|
||||
name: data.name,
|
||||
address: data.address,
|
||||
latitude: data.latitude,
|
||||
longitude: data.longitude,
|
||||
priceRegular: data.priceRegular,
|
||||
pricePremium: data.pricePremium,
|
||||
priceDiesel: data.priceDiesel,
|
||||
rating: data.rating,
|
||||
photoReference: data.photoReference,
|
||||
};
|
||||
|
||||
await this.stationsRepository.cacheStation(station);
|
||||
|
||||
// Get the created station
|
||||
const created = await this.stationsRepository.getCachedStation(data.placeId);
|
||||
if (!created) {
|
||||
throw new Error('Failed to retrieve created station');
|
||||
}
|
||||
|
||||
// Invalidate caches
|
||||
await this.invalidateStationCaches();
|
||||
|
||||
// Log audit action
|
||||
await this.adminRepository.logAuditAction(
|
||||
actorId,
|
||||
'CREATE',
|
||||
undefined,
|
||||
'station',
|
||||
data.placeId,
|
||||
{ name: data.name, address: data.address }
|
||||
);
|
||||
|
||||
logger.info('Station created by admin', { actorId, placeId: data.placeId });
|
||||
return created;
|
||||
} catch (error) {
|
||||
logger.error('Error creating station', { error, data });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update an existing station
|
||||
*/
|
||||
async updateStation(
|
||||
actorId: string,
|
||||
stationId: string,
|
||||
data: UpdateStationData
|
||||
): Promise<Station> {
|
||||
try {
|
||||
// First verify station exists
|
||||
const existing = await this.stationsRepository.getCachedStation(stationId);
|
||||
if (!existing) {
|
||||
throw new Error('Station not found');
|
||||
}
|
||||
|
||||
// Build update query dynamically based on provided fields
|
||||
const updates: string[] = [];
|
||||
const values: any[] = [];
|
||||
let paramIndex = 1;
|
||||
|
||||
if (data.name !== undefined) {
|
||||
updates.push(`name = $${paramIndex++}`);
|
||||
values.push(data.name);
|
||||
}
|
||||
if (data.address !== undefined) {
|
||||
updates.push(`address = $${paramIndex++}`);
|
||||
values.push(data.address);
|
||||
}
|
||||
if (data.latitude !== undefined) {
|
||||
updates.push(`latitude = $${paramIndex++}`);
|
||||
values.push(data.latitude);
|
||||
}
|
||||
if (data.longitude !== undefined) {
|
||||
updates.push(`longitude = $${paramIndex++}`);
|
||||
values.push(data.longitude);
|
||||
}
|
||||
if (data.priceRegular !== undefined) {
|
||||
updates.push(`price_regular = $${paramIndex++}`);
|
||||
values.push(data.priceRegular);
|
||||
}
|
||||
if (data.pricePremium !== undefined) {
|
||||
updates.push(`price_premium = $${paramIndex++}`);
|
||||
values.push(data.pricePremium);
|
||||
}
|
||||
if (data.priceDiesel !== undefined) {
|
||||
updates.push(`price_diesel = $${paramIndex++}`);
|
||||
values.push(data.priceDiesel);
|
||||
}
|
||||
if (data.rating !== undefined) {
|
||||
updates.push(`rating = $${paramIndex++}`);
|
||||
values.push(data.rating);
|
||||
}
|
||||
if (data.photoReference !== undefined) {
|
||||
updates.push(`photo_reference = $${paramIndex++}`);
|
||||
values.push(data.photoReference);
|
||||
}
|
||||
|
||||
if (updates.length === 0) {
|
||||
throw new Error('No fields to update');
|
||||
}
|
||||
|
||||
updates.push(`cached_at = NOW()`);
|
||||
values.push(stationId);
|
||||
|
||||
const query = `
|
||||
UPDATE station_cache
|
||||
SET ${updates.join(', ')}
|
||||
WHERE place_id = $${paramIndex}
|
||||
`;
|
||||
|
||||
await this.pool.query(query, values);
|
||||
|
||||
// Get updated station
|
||||
const updated = await this.stationsRepository.getCachedStation(stationId);
|
||||
if (!updated) {
|
||||
throw new Error('Failed to retrieve updated station');
|
||||
}
|
||||
|
||||
// Invalidate caches
|
||||
await this.invalidateStationCaches(stationId);
|
||||
|
||||
// Log audit action
|
||||
await this.adminRepository.logAuditAction(
|
||||
actorId,
|
||||
'UPDATE',
|
||||
undefined,
|
||||
'station',
|
||||
stationId,
|
||||
data
|
||||
);
|
||||
|
||||
logger.info('Station updated by admin', { actorId, stationId });
|
||||
return updated;
|
||||
} catch (error) {
|
||||
logger.error('Error updating station', { error, stationId, data });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a station (soft delete by default, hard delete with force flag)
|
||||
*/
|
||||
async deleteStation(
|
||||
actorId: string,
|
||||
stationId: string,
|
||||
force: boolean = false
|
||||
): Promise<void> {
|
||||
try {
|
||||
// Verify station exists
|
||||
const existing = await this.stationsRepository.getCachedStation(stationId);
|
||||
if (!existing) {
|
||||
throw new Error('Station not found');
|
||||
}
|
||||
|
||||
if (force) {
|
||||
// Hard delete - remove from both tables
|
||||
await this.pool.query('DELETE FROM station_cache WHERE place_id = $1', [stationId]);
|
||||
await this.pool.query('DELETE FROM saved_stations WHERE place_id = $1', [stationId]);
|
||||
|
||||
logger.info('Station hard deleted by admin', { actorId, stationId });
|
||||
} else {
|
||||
// Soft delete - add deleted_at column if not exists, then set it
|
||||
// First check if column exists
|
||||
const columnCheck = await this.pool.query(`
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'station_cache' AND column_name = 'deleted_at'
|
||||
`);
|
||||
|
||||
if (columnCheck.rows.length === 0) {
|
||||
// Add deleted_at column
|
||||
await this.pool.query(`
|
||||
ALTER TABLE station_cache
|
||||
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE
|
||||
`);
|
||||
}
|
||||
|
||||
// Soft delete
|
||||
await this.pool.query(
|
||||
'UPDATE station_cache SET deleted_at = NOW() WHERE place_id = $1',
|
||||
[stationId]
|
||||
);
|
||||
|
||||
logger.info('Station soft deleted by admin', { actorId, stationId });
|
||||
}
|
||||
|
||||
// Invalidate caches
|
||||
await this.invalidateStationCaches(stationId);
|
||||
|
||||
// Log audit action
|
||||
await this.adminRepository.logAuditAction(
|
||||
actorId,
|
||||
'DELETE',
|
||||
undefined,
|
||||
'station',
|
||||
stationId,
|
||||
{ force }
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('Error deleting station', { error, stationId, force });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get user's saved stations
|
||||
*/
|
||||
async getUserSavedStations(userId: string): Promise<SavedStation[]> {
|
||||
try {
|
||||
const stations = await this.stationsRepository.getUserSavedStations(userId);
|
||||
return stations;
|
||||
} catch (error) {
|
||||
logger.error('Error getting user saved stations', { error, userId });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove user's saved station (soft delete by default, hard delete with force)
|
||||
*/
|
||||
async removeUserSavedStation(
|
||||
actorId: string,
|
||||
userId: string,
|
||||
stationId: string,
|
||||
force: boolean = false
|
||||
): Promise<void> {
|
||||
try {
|
||||
if (force) {
|
||||
// Hard delete
|
||||
const result = await this.pool.query(
|
||||
'DELETE FROM saved_stations WHERE user_id = $1 AND place_id = $2',
|
||||
[userId, stationId]
|
||||
);
|
||||
|
||||
if ((result.rowCount ?? 0) === 0) {
|
||||
throw new Error('Saved station not found');
|
||||
}
|
||||
|
||||
logger.info('User saved station hard deleted by admin', { actorId, userId, stationId });
|
||||
} else {
|
||||
// Soft delete - add deleted_at column if not exists
|
||||
const columnCheck = await this.pool.query(`
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'saved_stations' AND column_name = 'deleted_at'
|
||||
`);
|
||||
|
||||
if (columnCheck.rows.length === 0) {
|
||||
// Column already exists in migration, but double check
|
||||
logger.warn('deleted_at column check executed', { table: 'saved_stations' });
|
||||
}
|
||||
|
||||
// Soft delete
|
||||
const result = await this.pool.query(
|
||||
'UPDATE saved_stations SET deleted_at = NOW() WHERE user_id = $1 AND place_id = $2 AND deleted_at IS NULL',
|
||||
[userId, stationId]
|
||||
);
|
||||
|
||||
if ((result.rowCount ?? 0) === 0) {
|
||||
throw new Error('Saved station not found or already deleted');
|
||||
}
|
||||
|
||||
logger.info('User saved station soft deleted by admin', { actorId, userId, stationId });
|
||||
}
|
||||
|
||||
// Invalidate user's saved stations cache
|
||||
await redis.del(`mvp:stations:saved:${userId}`);
|
||||
|
||||
// Log audit action
|
||||
await this.adminRepository.logAuditAction(
|
||||
actorId,
|
||||
'DELETE',
|
||||
undefined,
|
||||
'saved_station',
|
||||
`${userId}:${stationId}`,
|
||||
{ userId, stationId, force }
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('Error removing user saved station', { error, userId, stationId, force });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Invalidate station-related Redis caches
|
||||
*/
|
||||
private async invalidateStationCaches(stationId?: string): Promise<void> {
|
||||
try {
|
||||
// Get all keys matching station cache patterns
|
||||
const patterns = [
|
||||
'mvp:stations:*',
|
||||
'mvp:stations:search:*',
|
||||
];
|
||||
|
||||
for (const pattern of patterns) {
|
||||
const keys = await redis.keys(pattern);
|
||||
if (keys.length > 0) {
|
||||
await redis.del(...keys);
|
||||
}
|
||||
}
|
||||
|
||||
logger.info('Station caches invalidated', { stationId });
|
||||
} catch (error) {
|
||||
logger.error('Error invalidating station caches', { error, stationId });
|
||||
// Don't throw - cache invalidation failure shouldn't fail the operation
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Map database row to Station object
|
||||
*/
|
||||
private mapStationRow(row: any): Station {
|
||||
return {
|
||||
id: row.id,
|
||||
placeId: row.place_id,
|
||||
name: row.name,
|
||||
address: row.address,
|
||||
latitude: parseFloat(row.latitude),
|
||||
longitude: parseFloat(row.longitude),
|
||||
priceRegular: row.price_regular ? parseFloat(row.price_regular) : undefined,
|
||||
pricePremium: row.price_premium ? parseFloat(row.price_premium) : undefined,
|
||||
priceDiesel: row.price_diesel ? parseFloat(row.price_diesel) : undefined,
|
||||
rating: row.rating ? parseFloat(row.rating) : undefined,
|
||||
photoReference: row.photo_reference,
|
||||
lastUpdated: row.cached_at,
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user