cmd/generator/sync_generator.js - main - iezhelev/cppsh_micro

cmd/generator/sync_generator.js
209 строк8 KB
iezhelev
cdc message dispatcher (not ready) + ws client registry
03 апр 2025, 14:09
03 апр 2025, 14:095ed3279
const tconfs = require('./tables_configs.js')
 
function generateGoStruct(table, structName) {
  let structCode = `type ${structName} struct {\n`;
  table.fields.forEach(field => {
    let goType;
    // Extract the base type (e.g., "varchar(255)" -> "varchar")
    const baseType = field.type.split('(')[0].toLowerCase();

    switch (baseType) {
      case "int":
      case "integer":
      case "int4":
        goType = "int";
        break;
      case "bigint":
      case "int8":
        goType = "int64";
        break;
      case "smallint":
      case "int2":
        goType = "int16";
        break;
      case "varchar":
      case "text":
      case "char":
      case "bpchar":
      case "string":
        goType = "string";
        break;
      case "boolean":
      case "bool":
        goType = "bool";
        break;
      case "numeric":
      case "decimal":
      case "float4":
      case "float8":
        goType = "float64";
        break;
      case "timestamp":
      case "timestamptz":
      case "date":
      case "time":
        goType = "time.Time";
        break;
      case "json":
      case "jsonb":
        goType = "json.RawMessage";
        break;
      case "uuid":
        goType = "uuid.UUID";
        break;
      default:
        goType = "interface{}"; // Fallback for unknown types
    }

    // Handle nullable fields using sql.Null* types
    if (field.is_nullable) {
      switch (goType) {
        case "int":
          goType = "sql.NullInt64";
          break;
        case "int64":
          goType = "sql.NullInt64";
          break;
        case "string":
          goType = "sql.NullString";
          break;
        case "bool":
          goType = "sql.NullBool";
          break;
        case "float64":
          goType = "sql.NullFloat64";
          break;
        case "time.Time":
          goType = "sql.NullTime";
          break;
        default:
          goType = "sql.NullString"; // Fallback for nullable unknown types
      }
    }

    // Capitalize the first letter of the field name to make it exported
    const fieldName = field.name //.charAt(0).toUpperCase() + field.name.slice(1);
    // Add JSON struct tags for better serialization
    const jsonTag = field.name.toLowerCase();
    //structCode += `    ${fieldName} ${goType} \`json:"${jsonTag}"\`\n`;
    structCode += `    ${fieldName} ${goType}\n`;
  });
  structCode += "}\n";
  return structCode;
}

  function generateSyncFunction(sourceTable, destinationTable, fieldsMapper) {
    let syncCode = `func syncData_${destinationTable.name}(ctx context.Context, mainDB, dockerDB *sql.DB, rabbitConn *amqp.Connection, lastSyncTime time.Time) (time.Time, error) {\n`;
    syncCode += `    now := time.Now().UTC()\n`; // Capture the current time
    syncCode += `    rows, err := mainDB.QueryContext(ctx, \`SELECT `;
  
    sourceTable.fields.forEach((field, index) => {
      syncCode += `${field.name}`;
      if (index < sourceTable.fields.length - 1) {
        syncCode += ", ";
      }
    });
  
    syncCode += `\nFROM ${tconfs.schemas[sourceTable.name]}.${sourceTable.name}`;
    syncCode += `\nWHERE COALESCE(updatedtime, createdtime) > $1 AT TIME ZONE 'UTC'
    AND COALESCE(updatedtime, createdtime) <= $2 AT TIME ZONE 'UTC'
    ${ tconfs.tables_sync_select_conditions[sourceTable.name] || "" }\`, lastSyncTime, now)\n`; // Use now as the upper limit
    syncCode += `    if err != nil {\n`;
    syncCode += `        return lastSyncTime, fmt.Errorf("error fetching data: %w", err)\n`;
    syncCode += `    }\n`;
    syncCode += `    defer rows.Close()\n\n`;
    syncCode += `    var batchData []${destinationTable.name}\n`;
    // syncCode += `    var rabbitMessages []string\n\n`;
    syncCode += `    for rows.Next() {\n`;
    syncCode += `        var record ${destinationTable.name}\n`;
    syncCode += `        err := rows.Scan(\n`;
    destinationTable.fields.forEach(field => {
      syncCode += `            &record.${field.name},\n`;
    });
    syncCode += `        )\n`;
    syncCode += `        if err != nil {\n`;
    syncCode += `            log.Printf("Error scanning row: %v", err)\n`;
    syncCode += `            continue\n`;
    syncCode += `        }\n\n`;
    syncCode += `        batchData = append(batchData, record)\n`;
    // syncCode += `        rabbitMessages = append(rabbitMessages, fmt.Sprintf("New data: %+v", record))\n`;
    syncCode += `    }\n\n`;
    syncCode += `    if err := rows.Err(); err != nil {\n`;
    syncCode += `        return lastSyncTime, fmt.Errorf("row processing error: %w", err)\n`;
    syncCode += `    }\n\n`;
    syncCode += `    if len(batchData) == 0 {\n`;
    syncCode += `        log.Println("No data to sync ${destinationTable.name}")\n`;
    syncCode += `        return lastSyncTime, nil\n`;
    syncCode += `    }\n\n`;
    syncCode += `    tx, err := dockerDB.BeginTx(ctx, nil)\n`;
    syncCode += `    if err != nil {\n`;
    syncCode += `        return lastSyncTime, fmt.Errorf("failed to begin transaction: %w", err)\n`;
    syncCode += `    }\n\n`;
    syncCode += `    stmt, err := tx.PrepareContext(ctx, "INSERT INTO ${destinationTable.name} (`;
  
    destinationTable.fields.forEach((field, index) => {
      syncCode += `${field.name}`;
      if (index < destinationTable.fields.length - 1) {
        syncCode += ", ";
      }
    });
    syncCode += ",synctime"
    syncCode += `) VALUES (`;
  
    destinationTable.fields.forEach((field, index) => {
      syncCode += `$${index + 1}`;
      if (index < destinationTable.fields.length - 1) {
        syncCode += ", ";
      }
    });
    syncCode += ",CURRENT_TIMESTAMP AT TIME ZONE 'UTC'";
  
    syncCode += `) ON CONFLICT (${tconfs.primkeys[destinationTable.name]||(destinationTable.name+"id")}) DO UPDATE SET `;
  
    destinationTable.fields.forEach((field, index) => {
      syncCode += `${field.name} = EXCLUDED.${field.name}`;
      if (index < destinationTable.fields.length - 1) {
        syncCode += ", ";
      }
    });
    syncCode += ",synctime=CURRENT_TIMESTAMP AT TIME ZONE 'UTC'";
  
    syncCode += `")\n`;
    syncCode += `    if err != nil {\n`;
    syncCode += `        tx.Rollback()\n`;
    syncCode += `        return lastSyncTime, fmt.Errorf("failed to prepare statement: %w", err)\n`;
    syncCode += `    }\n`;
    syncCode += `    defer stmt.Close()\n\n`;
    syncCode += `    for _, record := range batchData {\n`;
    syncCode += `        _, err := stmt.Exec(\n`;
    destinationTable.fields.forEach(field => {
      syncCode += `            record.${fieldsMapper.fields_map[field.name] || field.name},\n`;
    });
    syncCode += `        )\n`;
    syncCode += `        if err != nil {\n`;
    syncCode += `            tx.Rollback()\n`;
    syncCode += `            return lastSyncTime, fmt.Errorf("failed to execute statement: %w", err)\n`;
    syncCode += `        }\n`;
    syncCode += `    }\n\n`;
    syncCode += `    if err := tx.Commit(); err != nil {\n`;
    syncCode += `        return lastSyncTime, fmt.Errorf("failed to commit transaction: %w", err)\n`;
    syncCode += `    }\n\n`;
    // syncCode += `    if err := PublishMessages(rabbitConn, "your_exchange", "your_routing_key", rabbitMessages); err != nil {\n`;
    // syncCode += `        return lastSyncTime, fmt.Errorf("error publishing messages: %w", err)\n`;
    // syncCode += `    }\n\n`;
    syncCode += `    log.Printf("Synced %d records ${destinationTable.name}", len(batchData))\n`;
    syncCode += `    lastSyncTime = now\n`;
    syncCode += `    return lastSyncTime, nil\n`;
    syncCode += `}\n`;
  
    return syncCode;
  }
  

  

  module.exports = {
    generateGoStruct: generateGoStruct,
    generateSyncFunction: generateSyncFunction,
    
};