queryer Reader,

	q db.Querier,

	if err := validateMappedSpecies(queryer, mappedSpeciesSet, &result); err != nil {

	if err := validateMappedSpecies(q, mappedSpeciesSet, &result); err != nil {

	if err := validateMappedCalltypes(queryer, mappedCalltypes, &result); err != nil {

	if err := validateMappedCalltypes(q, mappedCalltypes, &result); err != nil {

func validateMappedSpecies(queryer Reader, mappedSpeciesSet map[string]bool, result *mapping.ValidationResult) error {

func validateMappedSpecies(q db.Querier, mappedSpeciesSet map[string]bool, result *mapping.ValidationResult) error {

	rows, err := queryer.QueryContext(context.Background(), query, args...)

	rows, err := q.QueryContext(context.Background(), query, args...)

func validateMappedCalltypes(queryer Reader, mappedCalltypes map[string]map[string]string, result *mapping.ValidationResult) error {

func validateMappedCalltypes(q db.Querier, mappedCalltypes map[string]map[string]string, result *mapping.ValidationResult) error {

		rows, err := queryer.QueryContext(context.Background(), query, args...)

		rows, err := q.QueryContext(context.Background(), query, args...)

	database *sql.DB,

	q db.Querier,

	if err := validateSegmentHierarchy(database, input.DatasetID, input.LocationID, input.ClusterID); err != nil {

	if err := validateSegmentHierarchy(q, input.DatasetID, input.LocationID, input.ClusterID); err != nil {

	filterIDMap, err := validateFiltersExist(database, uniqueFilters)

	filterIDMap, err := validateFiltersExist(q, uniqueFilters)

	validationResult, err := ValidateMappingAgainstDB(database, mapping, uniqueSpecies, uniqueCalltypes)

	validationResult, err := ValidateMappingAgainstDB(q, mapping, uniqueSpecies, uniqueCalltypes)

	speciesIDMap, calltypeIDMap, err := loadSpeciesCalltypeIDs(database, mapping, uniqueSpecies, uniqueCalltypes)

	speciesIDMap, calltypeIDMap, err := loadSpeciesCalltypeIDs(q, mapping, uniqueSpecies, uniqueCalltypes)

	fileIDMap, hashErrors := validateAndMapFiles(database, scannedFiles, input.ClusterID, input.DatasetID)

	fileIDMap, hashErrors := validateAndMapFiles(q, scannedFiles, input.ClusterID, input.DatasetID)

	"database/sql"

	"skraak/db"

	tx *db.LoggedTx,

	m Mutator,

	if err := insertLabel(ctx, tx, ids, segmentID, label); err != nil {

	if err := insertLabel(ctx, m, ids, segmentID, label); err != nil {

		if err := insertLabelMetadata(ctx, tx, ids.labelID, label.Comment); err != nil {

		if err := insertLabelMetadata(ctx, m, ids.labelID, label.Comment); err != nil {

		if ctErr := importCalltype(ctx, tx, ids.labelID, label, ids.dbSpecies, ids.filterID, mapping, calltypeIDMap, sf); ctErr != nil {

		if ctErr := importCalltype(ctx, m, ids.labelID, label, ids.dbSpecies, ids.filterID, mapping, calltypeIDMap, sf); ctErr != nil {

func insertLabel(ctx context.Context, tx *db.LoggedTx, ids resolvedLabelIDs, segmentID string, label *datafile.Label) error {
	_, err := tx.ExecContext(ctx, `

func insertLabel(ctx context.Context, m Mutator, ids resolvedLabelIDs, segmentID string, label *datafile.Label) error {
	_, err := m.ExecContext(ctx, `

func insertLabelMetadata(ctx context.Context, tx *db.LoggedTx, labelID, comment string) error {

func insertLabelMetadata(ctx context.Context, m Mutator, labelID, comment string) error {

	_, err := tx.ExecContext(ctx, `

	_, err := m.ExecContext(ctx, `

	tx *db.LoggedTx,

	m Mutator,

	_, err = tx.ExecContext(ctx, `

	_, err = m.ExecContext(ctx, `

// importSegmentsIntoDB performs the transactional import

// importSegmentsIntoDB performs the transactional import using the provided mutator.

	database *sql.DB,

	m Mutator,

	tx, err := db.BeginLoggedTx(ctx, database, "import_segments")
	if err != nil {
		errors = append(errors, ImportSegmentError{
			Stage:   StageImport,
			Message: fmt.Sprintf("failed to begin transaction: %v", err),
		})
		return nil, 0, 0, nil, errors
	}
	defer tx.Rollback()

			segImp, labelIDs, subtypes, segErrs := importSegment(ctx, tx, seg, segIdx, sf, datasetID, mapping, filterIDMap, speciesIDMap, calltypeIDMap)

			segImp, labelIDs, subtypes, segErrs := importSegment(ctx, m, seg, segIdx, sf, datasetID, mapping, filterIDMap, speciesIDMap, calltypeIDMap)

				if _, err := tx.ExecContext(ctx, `DELETE FROM segment WHERE id = ?`, segImp.SegmentID); err != nil {

				if _, err := m.ExecContext(ctx, `DELETE FROM segment WHERE id = ?`, segImp.SegmentID); err != nil {

	if err := tx.Commit(); err != nil {
		errors = append(errors, ImportSegmentError{
			Stage:   StageImport,
			Message: fmt.Sprintf("failed to commit transaction: %v", err),
		})
		return nil, 0, 0, nil, errors
	}

	tx *db.LoggedTx,

	m Mutator,

	_, err = tx.ExecContext(ctx, `

	_, err = m.ExecContext(ctx, `

		result := importSingleLabel(ctx, tx, label, segmentID, segIdx, labelIdx, sf, mapping, filterIDMap, speciesIDMap, calltypeIDMap)

		result := importSingleLabel(ctx, m, label, segmentID, segIdx, labelIdx, sf, mapping, filterIDMap, speciesIDMap, calltypeIDMap)

	"database/sql"

	// Phase B+C: Parse data files and validate against DB
	database, err := db.OpenWriteableDB(db.ResolveDBPath(input.DBPath, ""))

	// Phase B+C: Parse data files and validate against DB (read-only)
	var val *segmentValidation
	err = db.WithReadDB(db.ResolveDBPath(input.DBPath, ""), func(database *sql.DB) error {
		var valErrors []ImportSegmentError
		var err error
		val, valErrors, err = validateAndPrepareSegments(database, input, mapping, dataFiles)
		output.Errors = append(output.Errors, valErrors...)
		return err
	})

		return output, fmt.Errorf("failed to open database: %w", err)
	}
	defer database.Close()
	val, valErrors, err := validateAndPrepareSegments(database, input, mapping, dataFiles)
	output.Errors = append(output.Errors, valErrors...)
	if err != nil {

	importedSegments, importedLabels, importedSubtypes, fileUpdates, importErrors := importSegmentsIntoDB(
		ctx, database, val.fileIDMap, val.scannedFiles, mapping, val.filterIDMap, val.speciesIDMap, val.calltypeIDMap, input.DatasetID, input.ProgressHandler,
	)

	var importedSegments []SegmentImport
	var importedLabels, importedSubtypes int
	var fileUpdates []dataFileUpdate
	var importErrors []ImportSegmentError
	err = db.WithWriteTx(ctx, db.ResolveDBPath(input.DBPath, ""), "import_segments", func(_ *sql.DB, tx *db.LoggedTx) error {
		importedSegments, importedLabels, importedSubtypes, fileUpdates, importErrors = importSegmentsIntoDB(
			ctx, tx, val.fileIDMap, val.scannedFiles, mapping, val.filterIDMap, val.speciesIDMap, val.calltypeIDMap, input.DatasetID, input.ProgressHandler,
		)
		return nil
	})

	if err != nil {
		return output, err
	}

	// Phase 3: Open database connection (single connection for all DB operations)
	database, err := db.OpenWriteableDB(db.ResolveDBPath(input.DBPath, ""))
	if err != nil {
		return output, fmt.Errorf("database connection failed: %w", err)
	}
	defer database.Close()
	// Phase 4: Get location data for astronomical calculations
	locData, err := GetLocationData(database, input.LocationID)
	if err != nil {
		return output, fmt.Errorf("failed to get location data: %w", err)
	}

	// Phase 3: Get location data and process file metadata
	var locData *LocationData
	var result *wav.FileProcessingResult
	err = db.WithReadDB(db.ResolveDBPath(input.DBPath, ""), func(database *sql.DB) error {
		var err error
		locData, err = GetLocationData(database, input.LocationID)
		if err != nil {
			return fmt.Errorf("failed to get location data: %w", err)
		}

	// Phase 5: Process file metadata
	result, err := wav.ProcessSingleFile(input.FilePath, locData.Latitude, locData.Longitude, locData.TimezoneID, true)

		result, err = wav.ProcessSingleFile(input.FilePath, locData.Latitude, locData.Longitude, locData.TimezoneID, true)
		if err != nil {
			return fmt.Errorf("file processing failed: %w", err)
		}
		return nil
	})

		return output, fmt.Errorf("file processing failed: %w", err)

		return output, err

	// Phase 6: Insert into database (includes EnsureClusterPath)
	fileID, isDuplicate, err := insertFileIntoDB(ctx, database, result, input)

	// Phase 4: Insert into database (includes EnsureClusterPath)
	var fileID string
	var isDuplicate bool
	err = db.WithWriteTx(ctx, db.ResolveDBPath(input.DBPath, ""), "import_audio_file", func(_ *sql.DB, tx *db.LoggedTx) error {
		var err error
		fileID, isDuplicate, err = insertFileIntoDB(ctx, tx, result, input)
		return err
	})

// insertFileIntoDB inserts a single file into the database

// insertFileIntoDB inserts a single file into the database using the provided mutator.

	database *sql.DB,

	m Mutator,

	// Begin logged transaction
	tx, err := db.BeginLoggedTx(ctx, database, "import_audio_file")
	if err != nil {
		return "", false, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback() // Rollback if not committed
	// Ensure cluster path is set (inside transaction for logging + rollback safety)
	if err := EnsureClusterPath(tx, input.ClusterID, filepath.Dir(input.FilePath)); err != nil {

	// Ensure cluster path is set
	if err := EnsureClusterPath(m, input.ClusterID, filepath.Dir(input.FilePath)); err != nil {

	existingID, isDup, err := CheckDuplicateHash(tx, result.Hash)

	existingID, isDup, err := CheckDuplicateHash(m, result.Hash)

		// Commit the transaction to log any EnsureClusterPath UPDATE
		_ = tx.Commit()

	fileID, err := insertNewFileRecord(ctx, tx, result, input)

	fileID, err := insertNewFileRecord(ctx, m, result, input)

	}
	// Commit transaction
	if err = tx.Commit(); err != nil {
		return "", false, fmt.Errorf("transaction commit failed: %w", err)

	tx *db.LoggedTx,

	m Mutator,

	_, err = tx.ExecContext(ctx, `

	_, err = m.ExecContext(ctx, `

	_, err = tx.ExecContext(ctx, `

	_, err = m.ExecContext(ctx, `

		_, err = tx.ExecContext(ctx, `

		_, err = m.ExecContext(ctx, `

func bulkCreateClusters(ctx context.Context, database *sql.DB, logger *progressLogger, locations []bulkLocationData, datasetID string) (map[string]string, int, int, error) {

func bulkCreateClusters(ctx context.Context, m Mutator, logger *progressLogger, locations []bulkLocationData, datasetID string) (map[string]string, int, int, error) {

		err := database.QueryRowContext(context.Background(), `

		err := m.QueryRowContext(context.Background(), `

			clusterID, err = bulkCreateCluster(ctx, database, datasetID, loc.LocationID, loc.DateRange, loc.SampleRate)

			clusterID, err = bulkCreateCluster(ctx, m, datasetID, loc.LocationID, loc.DateRange, loc.SampleRate)

func bulkImportAllFiles(database *sql.DB, logger *progressLogger, locations []bulkLocationData, clusterIDMap map[string]string, datasetID string) (bulkImportStats, []string) {

func bulkImportAllFiles(m Mutator, logger *progressLogger, locations []bulkLocationData, clusterIDMap map[string]string, datasetID string) (bulkImportStats, []string) {

		stats, err := bulkImportFilesForCluster(database, logger, loc.DirectoryPath, datasetID, loc.LocationID, clusterID)

		stats, err := bulkImportFilesForCluster(m, logger, loc.DirectoryPath, datasetID, loc.LocationID, clusterID)

func bulkCreateCluster(ctx context.Context, database *sql.DB, datasetID, locationID, name string, sampleRate int) (string, error) {

func bulkCreateCluster(ctx context.Context, m Mutator, datasetID, locationID, name string, sampleRate int) (string, error) {

	err = database.QueryRowContext(context.Background(), "SELECT name FROM location WHERE id = ?", locationID).Scan(&locationName)

	err = m.QueryRowContext(context.Background(), "SELECT name FROM location WHERE id = ?", locationID).Scan(&locationName)

	tx, err := db.BeginLoggedTx(ctx, database, "bulk_file_import")
	if err != nil {
		return "", fmt.Errorf("failed to begin transaction: %w", err)

	// If m is a LoggedTx, use it directly; otherwise begin a new transaction
	var tx *db.LoggedTx
	switch v := m.(type) {
	case *db.LoggedTx:
		tx = v
	default:
		tx, err = db.BeginLoggedTx(ctx, v.(*sql.DB), "bulk_file_import")
		if err != nil {
			return "", fmt.Errorf("failed to begin transaction: %w", err)
		}
		defer tx.Rollback()

	defer tx.Rollback()

	if err = tx.Commit(); err != nil {
		return "", fmt.Errorf("failed to commit cluster creation: %w", err)

	// Only commit if we created the transaction
	if _, isLoggedTx := m.(*db.LoggedTx); !isLoggedTx {
		if err = tx.Commit(); err != nil {
			return "", fmt.Errorf("failed to commit cluster creation: %w", err)
		}

// resolveTransaction extracts transaction info from a Mutator.
// Returns the LoggedTx, whether caller should commit, and optionally a *sql.DB if available.
func resolveTransaction(ctx context.Context, m Mutator, txName string) (*db.LoggedTx, bool, *sql.DB, error) {
	switch v := m.(type) {
	case *db.LoggedTx:
		// Already in a transaction, use it
		return v, false, nil, nil
	case *sql.DB:
		// Need to create a transaction
		tx, err := db.BeginLoggedTx(ctx, v, txName)
		if err != nil {
			return nil, false, nil, fmt.Errorf("failed to begin transaction: %w", err)
		}
		return tx, true, v, nil
	default:
		return nil, false, nil, fmt.Errorf("unsupported Mutator type")
	}
}
// runImportCluster executes ImportCluster with appropriate Reader/Mutator.
func runImportCluster(database *sql.DB, tx *db.LoggedTx, input ClusterImportInput) (*ClusterImportOutput, error) {
	if database != nil {
		return ImportCluster(database, tx, input)
	}
	// tx is both Reader and Mutator
	return ImportCluster(tx, tx, input)
}

func bulkImportFilesForCluster(database *sql.DB, logger *progressLogger, folderPath, datasetID, locationID, clusterID string) (*bulkImportStats, error) {

func bulkImportFilesForCluster(m Mutator, logger *progressLogger, folderPath, datasetID, locationID, clusterID string) (*bulkImportStats, error) {

	tx, err := db.BeginLoggedTx(ctx, database, "import_audio_files")

	// Determine if we need to create a transaction or use existing one
	tx, shouldCommit, database, err := resolveTransaction(ctx, m, "import_audio_files")

		return nil, fmt.Errorf("failed to begin transaction: %w", err)

		return nil, err

		tx.Rollback()

		if shouldCommit {
			tx.Rollback()
		}

	clusterOutput, err := ImportCluster(database, tx, ClusterImportInput{

	// For ImportCluster, we need both Reader and Mutator
	clusterOutput, err := runImportCluster(database, tx, ClusterImportInput{

		tx.Rollback()

		if shouldCommit {
			tx.Rollback()
		}

	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("transaction commit failed: %w", err)

	if shouldCommit {
		if err := tx.Commit(); err != nil {
			return nil, fmt.Errorf("transaction commit failed: %w", err)
		}