16 KiB
16 KiB
Database Design and Migration Guide
Current Database Schema Analysis
Existing Collections
1. JobAssign Collection (Current)
{
_id: ObjectId,
job: Number, // Reference to Job._id
user: ObjectId, // Reference to User._id (aircraft/device)
status: Number, // 0: pending, 1: downloaded, 2: completed
date: Date
}
2. Job Collection (Current)
{
_id: Number,
name: String,
// ... existing fields
dlOp: {
type: Number,
mapOp: Mixed
}
// ... other fields
}
3. Application Collection (Current)
{
_id: ObjectId,
jobId: Number,
fileName: String,
fileSize: Number,
status: Number,
// ... other fields
}
Enhanced Schema Design
1. Enhanced JobAssign Schema
const JobAssignSchema = new Schema({
// Existing fields (unchanged for backward compatibility)
_id: ObjectId,
job: { type: Number, ref: 'Job', required: true },
user: { type: ObjectId, ref: 'User', required: true },
status: {
type: Number,
enum: [0, 1, 2], // 0: pending, 1: downloaded, 2: completed
default: 0
},
date: { type: Date, default: Date.now },
// New partner integration fields
partnerType: {
type: String,
enum: ['internal', 'satloc', 'dji', 'parrot', 'other'],
default: 'internal',
index: true
},
externalJobId: {
type: String,
sparse: true, // Only index non-null values
index: true
},
partnerMetadata: {
type: Schema.Types.Mixed,
default: null
},
// Sync state tracking for partner operations
syncState: {
jobUpload: {
status: {
type: String,
enum: ['pending', 'syncing', 'synced', 'failed'],
default: function() {
return this.partnerType === 'internal' ? 'synced' : 'pending';
}
},
attempts: { type: Number, default: 0, min: 0 },
lastAttempt: { type: Date },
lastSuccess: { type: Date },
error: { type: String },
errorCode: { type: String }, // For categorizing errors
nextRetry: { type: Date } // Scheduled next retry time
},
dataPolling: {
status: {
type: String,
enum: ['idle', 'polling', 'synced', 'failed'],
default: 'idle'
},
attempts: { type: Number, default: 0, min: 0 },
lastAttempt: { type: Date },
lastSuccess: { type: Date },
lastDataCheck: { type: Date },
error: { type: String },
errorCode: { type: String },
nextPoll: { type: Date }, // Scheduled next poll time
pollInterval: { type: Number, default: 60000 } // ms
}
},
// Partner-specific configuration
partnerConfig: {
aircraftId: { type: String }, // Partner's aircraft identifier
priority: {
type: String,
enum: ['low', 'normal', 'high'],
default: 'normal'
},
timeout: { type: Number, default: 30000 }, // Request timeout in ms
retryPolicy: {
maxAttempts: { type: Number, default: 5, min: 1, max: 10 },
baseDelay: { type: Number, default: 5000, min: 1000 },
maxDelay: { type: Number, default: 300000 },
backoffMultiplier: { type: Number, default: 2, min: 1, max: 5 },
jitter: { type: Boolean, default: true }
}
},
// Performance tracking
metrics: {
syncDuration: { type: Number }, // Total sync time in ms
dataSize: { type: Number }, // Amount of data processed
conversionTime: { type: Number }, // Time to convert data format
uploadTime: { type: Number }, // Time to upload to partner
downloadTime: { type: Number } // Time to download from partner
}
}, {
timestamps: true,
toJSON: { virtuals: true },
toObject: { virtuals: true }
});
// Compound indexes for efficient queries
JobAssignSchema.index({ user: 1, status: 1 });
JobAssignSchema.index({ job: 1, partnerType: 1 });
JobAssignSchema.index({ partnerType: 1, 'syncState.jobUpload.status': 1 });
JobAssignSchema.index({ partnerType: 1, 'syncState.dataPolling.status': 1 });
JobAssignSchema.index({ partnerType: 1, 'syncState.jobUpload.nextRetry': 1 });
JobAssignSchema.index({ partnerType: 1, 'syncState.dataPolling.nextPoll': 1 });
JobAssignSchema.index({ externalJobId: 1, partnerType: 1 }, { unique: true, sparse: true });
// Virtual fields
JobAssignSchema.virtual('isPartnerAssignment').get(function() {
return this.partnerType !== 'internal';
});
JobAssignSchema.virtual('needsSync').get(function() {
return this.isPartnerAssignment &&
this.syncState.jobUpload.status === 'pending';
});
JobAssignSchema.virtual('needsPolling').get(function() {
return this.isPartnerAssignment &&
this.syncState.jobUpload.status === 'synced' &&
this.syncState.dataPolling.status === 'idle' &&
this.status < 2;
});
2. Enhanced Application Schema
const EnhancedApplicationSchema = new Schema({
// Existing fields (unchanged)
jobId: { type: Number, required: true },
fileName: { type: String, required: true },
fileSize: { type: Number, required: true },
status: { type: Number, default: 1 },
// Enhanced fields for partner integration
partnerType: {
type: String,
enum: ['internal', 'satloc', 'dji', 'parrot', 'other'],
default: 'internal',
index: true
},
externalJobId: {
type: String,
sparse: true,
index: true
},
assignmentId: {
type: ObjectId,
ref: 'JobAssign',
index: true
},
// Original data format and metadata
originalData: {
format: {
type: String,
enum: ['agnav', 'satloc', 'dji_log', 'parrot_log', 'csv', 'other']
},
version: { type: String },
encoding: { type: String, default: 'utf8' },
compression: { type: String },
checksum: { type: String }
},
// Partner-specific metadata
partnerMetadata: {
aircraftInfo: {
id: String,
model: String,
firmware: String,
sensors: [String]
},
flightInfo: {
startTime: Date,
endTime: Date,
duration: Number,
altitude: { min: Number, max: Number, avg: Number },
speed: { min: Number, max: Number, avg: Number },
weather: Schema.Types.Mixed
},
dataQuality: {
gpsAccuracy: Number,
signalStrength: Number,
dataCompleteness: Number, // percentage
anomalies: [String]
}
},
// Processing pipeline tracking
processingStage: {
type: String,
enum: [
'uploaded', // File uploaded/received
'validated', // Initial validation complete
'parsing', // Currently parsing file
'converting', // Converting to internal format
'processing', // Processing application details
'completed', // Successfully processed
'failed' // Processing failed
],
default: 'uploaded',
index: true
},
processingSteps: [{
step: {
type: String,
enum: ['upload', 'validate', 'parse', 'convert', 'process', 'finalize']
},
status: {
type: String,
enum: ['pending', 'running', 'completed', 'failed', 'skipped']
},
startTime: Date,
endTime: Date,
duration: Number, // milliseconds
error: String,
metadata: Schema.Types.Mixed
}],
// Conversion and processing metrics
conversionMetrics: {
recordsInput: { type: Number, min: 0 },
recordsOutput: { type: Number, min: 0 },
dataLossPercentage: { type: Number, min: 0, max: 100 },
conversionTime: { type: Number }, // milliseconds
conversionErrors: [String],
formatMappings: Schema.Types.Mixed
},
// Quality assurance
qualityChecks: [{
checkType: {
type: String,
enum: ['gps_continuity', 'spray_coverage', 'data_integrity', 'format_compliance']
},
status: {
type: String,
enum: ['passed', 'failed', 'warning', 'skipped']
},
score: { type: Number, min: 0, max: 100 },
issues: [String],
recommendations: [String]
}],
// File storage information
storage: {
originalPath: String,
processedPath: String,
backupPath: String,
compressionRatio: Number,
storageSize: Number
}
}, {
timestamps: true,
toJSON: { virtuals: true },
toObject: { virtuals: true }
});
// Indexes for performance
EnhancedApplicationSchema.index({ jobId: 1, partnerType: 1 });
EnhancedApplicationSchema.index({ processingStage: 1, createdAt: 1 });
EnhancedApplicationSchema.index({ assignmentId: 1 });
EnhancedApplicationSchema.index({ externalJobId: 1, partnerType: 1 });
// Virtual fields
EnhancedApplicationSchema.virtual('isPartnerData').get(function() {
return this.partnerType !== 'internal';
});
EnhancedApplicationSchema.virtual('processingDuration').get(function() {
if (this.processingSteps && this.processingSteps.length > 0) {
const firstStep = this.processingSteps.find(s => s.startTime);
const lastStep = [...this.processingSteps].reverse().find(s => s.endTime);
if (firstStep && lastStep) {
return lastStep.endTime - firstStep.startTime;
}
}
return null;
});
3. New Partner Configuration Schema
const PartnerConfigSchema = new Schema({
code: {
type: String,
required: true,
unique: true,
lowercase: true,
match: /^[a-z0-9_]+$/
},
name: { type: String, required: true },
description: { type: String },
active: { type: Boolean, default: true },
// API Configuration
apiConfig: {
baseUrl: { type: String, required: true },
version: { type: String, default: 'v1' },
authentication: {
type: {
type: String,
enum: ['api_key', 'oauth2', 'basic_auth', 'bearer_token'],
required: true
},
credentials: Schema.Types.Mixed // Encrypted in production
},
rateLimit: {
requestsPerSecond: { type: Number, default: 10 },
burstLimit: { type: Number, default: 50 },
backoffStrategy: {
type: String,
enum: ['fixed', 'exponential', 'linear'],
default: 'exponential'
}
},
timeout: {
connect: { type: Number, default: 10000 },
request: { type: Number, default: 30000 },
response: { type: Number, default: 60000 }
}
},
// Capabilities and features
capabilities: [{
type: String,
enum: [
'job_upload',
'job_download',
'data_polling',
'real_time_sync',
'flight_planning',
'telemetry_streaming',
'weather_integration',
'obstacle_avoidance'
]
}],
// Data format specifications
dataFormats: {
input: [{
format: String,
version: String,
mimeType: String,
extensions: [String],
maxSize: Number,
compression: [String]
}],
output: [{
format: String,
version: String,
mimeType: String,
schema: Schema.Types.Mixed
}]
},
// Integration settings
integrationSettings: {
syncInterval: { type: Number, default: 60000 }, // ms
batchSize: { type: Number, default: 100 },
maxRetries: { type: Number, default: 3 },
healthCheckInterval: { type: Number, default: 300000 }, // 5 minutes
// Webhook configuration
webhooks: {
enabled: { type: Boolean, default: false },
endpoints: [{
event: {
type: String,
enum: ['job_completed', 'data_available', 'error_occurred', 'status_changed']
},
url: String,
secret: String,
retries: { type: Number, default: 3 }
}]
}
},
// Monitoring and alerting
monitoring: {
healthcheck: {
endpoint: String,
expectedResponse: Schema.Types.Mixed,
timeout: { type: Number, default: 10000 }
},
alerts: {
errorThreshold: { type: Number, default: 0.1 }, // 10% error rate
responseTimeThreshold: { type: Number, default: 5000 }, // 5 seconds
downtime: {
consecutiveFailures: { type: Number, default: 3 },
notificationChannels: [String]
}
}
}
}, {
timestamps: true
});
Migration Strategy
Phase 1: Backward Compatible Changes
-
Add new fields to existing schemas
// Add to JobAssign collection db.job_assigns.updateMany( { partnerType: { $exists: false } }, { $set: { partnerType: 'internal', syncState: { jobUpload: { status: 'synced', attempts: 0 }, dataPolling: { status: 'idle', attempts: 0 } } } } ); -
Create indexes gradually
// Create indexes in background db.job_assigns.createIndex( { partnerType: 1, "syncState.jobUpload.status": 1 }, { background: true } );
Phase 2: Data Migration Scripts
// Migration script for existing assignments
async function migrateExistingAssignments() {
const assignments = await JobAssign.find({
partnerType: { $exists: false }
});
for (const assignment of assignments) {
await JobAssign.findByIdAndUpdate(assignment._id, {
partnerType: 'internal',
syncState: {
jobUpload: {
status: 'synced',
attempts: 0,
lastSuccess: assignment.date
},
dataPolling: {
status: assignment.status > 1 ? 'synced' : 'idle',
attempts: 0
}
},
partnerConfig: {
retryPolicy: {
maxAttempts: 5,
baseDelay: 5000,
maxDelay: 300000,
backoffMultiplier: 2,
jitter: true
}
}
});
}
}
// Migration script for application records
async function migrateApplicationRecords() {
const apps = await Application.find({
partnerType: { $exists: false }
});
for (const app of apps) {
await Application.findByIdAndUpdate(app._id, {
partnerType: 'internal',
originalData: {
format: 'agnav',
version: '1.0',
encoding: 'utf8'
},
processingStage: app.status === 3 ? 'completed' : 'uploaded',
processingSteps: [{
step: 'upload',
status: 'completed',
startTime: app.createdAt,
endTime: app.createdAt,
duration: 0
}]
});
}
}
Phase 3: Validation and Testing
// Validation script to ensure data integrity
async function validateMigration() {
// Check all assignments have required fields
const invalidAssignments = await JobAssign.countDocuments({
$or: [
{ partnerType: { $exists: false } },
{ syncState: { $exists: false } }
]
});
if (invalidAssignments > 0) {
throw new Error(`${invalidAssignments} assignments missing required fields`);
}
// Check partner type consistency
const invalidPartnerTypes = await JobAssign.countDocuments({
partnerType: { $nin: ['internal', 'satloc', 'dji', 'parrot', 'other'] }
});
if (invalidPartnerTypes > 0) {
throw new Error(`${invalidPartnerTypes} assignments with invalid partner types`);
}
console.log('Migration validation passed');
}
Performance Considerations
Query Optimization
-
Compound Indexes
// For partner sync operations { partnerType: 1, "syncState.jobUpload.status": 1, "syncState.jobUpload.nextRetry": 1 } // For data polling { partnerType: 1, "syncState.dataPolling.nextPoll": 1, status: 1 } // For user queries { user: 1, status: 1, partnerType: 1 } -
Sparse Indexes
// Only index documents with external job IDs { externalJobId: 1 }, { sparse: true }
Data Archival Strategy
// Archive completed assignments older than 90 days
const archiveOldAssignments = async () => {
const cutoffDate = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
const oldAssignments = await JobAssign.find({
status: 2,
updatedAt: { $lt: cutoffDate }
});
// Move to archive collection
if (oldAssignments.length > 0) {
await JobAssignArchive.insertMany(oldAssignments);
await JobAssign.deleteMany({
_id: { $in: oldAssignments.map(a => a._id) }
});
}
console.log(`Archived ${oldAssignments.length} assignments`);
};
Backup and Recovery
Backup Strategy
- Full daily backups of partner configuration
- Incremental hourly backups of assignment data
- Point-in-time recovery for critical operations
- Cross-region replication for disaster recovery
Recovery Procedures
- Partner outage recovery: Automatic retry with exponential backoff
- Data corruption recovery: Restore from backup and replay operations
- Migration rollback: Automated rollback scripts for each phase
This enhanced database design provides a robust foundation for multi-partner integration while maintaining backward compatibility and performance.