crud.py

from datetime import datetime, timezone

from loguru import logger
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_fixed

from crowdgit.enums import RepositoryPriority, RepositoryState
from crowdgit.errors import RepoLockingError
from crowdgit.models.affiliation_info import RepoAffiliationRegistry
from crowdgit.models.repository import Repository
from crowdgit.models.service_execution import ServiceExecution
from crowdgit.settings import (
    FAILED_RETRY_INTERVAL_HOURS,
    MAX_CONCURRENT_ONBOARDINGS,
    MAX_INTEGRATION_RESULTS,
    REPOSITORY_UPDATE_INTERVAL_HOURS,
)

from .connection import get_db_connection
from .registry import execute, executemany, fetchrow, fetchval, query

# Common SELECT columns joining public.repositories + git.repositoryProcessing with aliases for backwards compatibility
REPO_SELECT_COLUMNS = """
    r.id,
    r.url,
    r."segmentId",
    r."gitIntegrationId",
    r."forkedFrom",
    rp.state,
    rp.priority,
    rp."lockedAt",
    rp."lastProcessedAt",
    rp."lastProcessedCommit",
    rp.branch,
    rp."maintainerFile",
    rp."lastMaintainerRunAt",
    rp."reOnboardingCount"
"""


async def get_recently_processed_repository_by_url(url: str) -> Repository | None:
    """
    Get repository by URL that was processed within the configured update interval.

    Returns the repository only if it was last processed within REPOSITORY_UPDATE_INTERVAL_HOURS
    and has a COMPLETED state.
    Used to check if a repository needs reprocessing based on the update interval.
    """
    sql_query = f"""
    SELECT {REPO_SELECT_COLUMNS}
    FROM public.repositories r
    JOIN git."repositoryProcessing" rp ON rp."repositoryId" = r.id
    WHERE r.url = $1
        AND r."deletedAt" IS NULL
        AND rp."lastProcessedAt" > NOW() - INTERVAL '1 hour' * $2
        AND rp.state = $3
    """
    result = await fetchrow(
        sql_query, (url, REPOSITORY_UPDATE_INTERVAL_HOURS, RepositoryState.COMPLETED)
    )
    return Repository.from_db(dict(result)) if result else None


async def acquire_onboarding_repo() -> Repository | None:
    onboarding_repo_sql_query = f"""
    WITH current_onboarding_count AS (
        SELECT COUNT(*) as count
        FROM git."repositoryProcessing" rp
        JOIN public.repositories r ON r.id = rp."repositoryId"
        WHERE rp.state = $1
            AND rp."lastProcessedCommit" IS NULL
            AND r."deletedAt" IS NULL
    ),
    selected_repo AS (
        SELECT r.id
        FROM public.repositories r
        JOIN git."repositoryProcessing" rp ON rp."repositoryId" = r.id
        CROSS JOIN current_onboarding_count c
        WHERE rp.state = $2
            AND rp."lockedAt" IS NULL
            AND r."deletedAt" IS NULL
            AND c.count < $3
        ORDER BY rp.priority ASC, rp."createdAt" ASC
        LIMIT 1
        FOR UPDATE OF rp SKIP LOCKED
    )
    UPDATE git."repositoryProcessing" rp
    SET "lockedAt" = NOW(),
        state = $1,
        "updatedAt" = NOW()
    FROM public.repositories r
    CROSS JOIN selected_repo
    WHERE rp."repositoryId" = r.id
        AND rp."repositoryId" = selected_repo.id
    RETURNING {REPO_SELECT_COLUMNS}
    """
    return await acquire_repository(
        onboarding_repo_sql_query,
        (RepositoryState.PROCESSING, RepositoryState.PENDING, MAX_CONCURRENT_ONBOARDINGS),
    )


@retry(
    retry=retry_if_exception_type(RepoLockingError),
    stop=stop_after_attempt(3),
    wait=wait_fixed(1),
    reraise=True,
)
async def acquire_repository(query: str, params: tuple = None) -> Repository | None:
    async with get_db_connection() as conn:
        try:
            async with conn.transaction():
                result = await conn.fetchrow(query, *params)
                if result:
                    repo = Repository.from_db(dict(result))
                    logger.info(f"Acquired repository: {repo.url}")
                    return repo
                # logger.info(
                #     "No repository is available for processing based on the filtering rules"
                # )
                return None
        except Exception as e:
            logger.error(f"failed to acquire repository with error: {e}. Retrying...")
            raise RepoLockingError() from e


async def acquire_recurrent_repo() -> Repository | None:
    """Acquire a regular (non-onboarding) repository, that were not processed in the last x hours (REPOSITORY_UPDATE_INTERVAL_HOURS)"""
    recurrent_repo_sql_query = f"""
    -- Rate-limit guard: Gerrit (automotivelinux) aggressively rate-limits concurrent connections.
    -- This CTE checks if any automotivelinux repo is already being processed,
    -- so we skip picking another one from the same host until the current one finishes.
    WITH automotivelinux_processing AS (
        SELECT 1
        FROM git."repositoryProcessing" rp2
        JOIN public.repositories r2 ON r2.id = rp2."repositoryId"
        WHERE rp2.state = 'processing'
            AND rp2."lockedAt" IS NOT NULL
            AND r2.url LIKE '%gerrit.automotivelinux.org%'
        LIMIT 1
    ),
    selected_repo AS (
        SELECT r.id
        FROM public.repositories r
        JOIN git."repositoryProcessing" rp ON rp."repositoryId" = r.id
        WHERE NOT (rp.state = ANY($2))
            AND rp."lockedAt" IS NULL
            AND r."deletedAt" IS NULL
            AND rp."lastProcessedAt" < NOW() - INTERVAL '1 hour' * (
                CASE WHEN rp.state = 'failed' THEN $4::numeric ELSE $3::numeric END
            )
            AND NOT (
                r.url LIKE '%gerrit.automotivelinux.org%'
                AND EXISTS (SELECT 1 FROM automotivelinux_processing)
            )
        ORDER BY rp.priority ASC, rp."lastProcessedAt" ASC
        LIMIT 1
        FOR UPDATE OF rp SKIP LOCKED
    )
    UPDATE git."repositoryProcessing" rp
    SET "lockedAt" = NOW(),
        state = $1,
        "updatedAt" = NOW()
    FROM public.repositories r
    CROSS JOIN selected_repo
    WHERE rp."repositoryId" = r.id
        AND rp."repositoryId" = selected_repo.id
    RETURNING {REPO_SELECT_COLUMNS}
    """
    states_to_exclude = (
        RepositoryState.PENDING,
        RepositoryState.PROCESSING,
        RepositoryState.PENDING_REONBOARD,
        RepositoryState.AUTH_REQUIRED,
    )
    return await acquire_repository(
        recurrent_repo_sql_query,
        (
            RepositoryState.PROCESSING,
            states_to_exclude,
            REPOSITORY_UPDATE_INTERVAL_HOURS,
            FAILED_RETRY_INTERVAL_HOURS,
        ),
    )


async def can_onboard_more():
    """
    Check if system can handle more repository onboarding based on activity load.

    Returns False if integration.results count exceeds MAX_INTEGRATION_RESULTS
    or if the query fails (indicating high database load).
    """
    try:
        integration_results_count = await fetchval("SELECT COUNT(*) FROM integration.results")
        return integration_results_count < MAX_INTEGRATION_RESULTS
    except Exception as e:
        logger.warning(f"Failed to get integration.results count with error: {repr(e)}")
        return False  # if query failed mostly due to timeout then db is already under high load


async def acquire_pending_reonboard_repo() -> Repository | None:
    """Acquire a pending_reonboard repo for re-onboarding (only called on weekends)."""
    pending_reonboard_sql_query = f"""
    WITH selected_repo AS (
        SELECT r.id
        FROM public.repositories r
        JOIN git."repositoryProcessing" rp ON rp."repositoryId" = r.id
        WHERE rp.state = $1
            AND rp."lockedAt" IS NULL
            AND r."deletedAt" IS NULL
        ORDER BY rp.priority ASC, rp."lastProcessedAt" ASC
        LIMIT 1
        FOR UPDATE OF rp SKIP LOCKED
    )
    UPDATE git."repositoryProcessing" rp
    SET "lockedAt" = NOW(),
        state = $2,
        "lastProcessedCommit" = NULL,
        branch = NULL,
        "reOnboardingCount" = rp."reOnboardingCount" + 1,
        "updatedAt" = NOW()
    FROM public.repositories r
    CROSS JOIN selected_repo
    WHERE rp."repositoryId" = r.id
        AND rp."repositoryId" = selected_repo.id
    RETURNING {REPO_SELECT_COLUMNS}
    """
    return await acquire_repository(
        pending_reonboard_sql_query,
        (RepositoryState.PENDING_REONBOARD, RepositoryState.PROCESSING),
    )


async def acquire_repo_for_processing() -> Repository | None:
    """
    Acquire the next repository to process based on priority and system load.

    Priority logic:
    1. Onboarding repos (PENDING state) - only if system load allows and
       current onboarding count is below MAX_CONCURRENT_ONBOARDINGS
    2. Recurrent repos (non-PENDING/non-PROCESSING) - fallback when onboarding
       is unavailable or skipped due to high load
    3. Pending reonboard repos (PENDING_REONBOARD state) - weekend-only, lowest priority.
       These are repos needing re-onboarding that were deferred until the weekend.

    Onboarding is delayed when integration.results exceeds MAX_INTEGRATION_RESULTS
    to prevent overloading the system during high activity periods.
    """
    repo_to_process = None
    if await can_onboard_more():
        repo_to_process = await acquire_onboarding_repo()
    else:
        logger.info("Skipping onboarding due to high load on integration.results")

    if not repo_to_process:
        is_weekend = datetime.now(timezone.utc).weekday() >= 5
        if is_weekend:
            repo_to_process = await acquire_pending_reonboard_repo()

    if not repo_to_process:
        repo_to_process = await acquire_recurrent_repo()

    return repo_to_process


async def release_repo(repo_id: str):
    """
    Release repository lock (lockedAt) after processing
    """
    sql_query = """
    UPDATE git."repositoryProcessing"
        SET "lockedAt" = NULL,
        "updatedAt" = NOW()
    WHERE "repositoryId" = $1
    """
    result = await execute(sql_query, (repo_id,))
    return str(result)


async def update_last_processed_commit(repo_id: str, commit_hash: str, branch: str | None = None):
    """
    Update last processed commit and optionally the branch after processing
    """
    sql_query = """
    UPDATE git."repositoryProcessing"
        SET "lastProcessedCommit" = $1,
        "branch" = $2,
        "updatedAt" = NOW()
    WHERE "repositoryId" = $3
    """
    result = await execute(sql_query, (commit_hash, branch, repo_id))
    return str(result)


async def update_repository_licenses(repository_id: str, licenses: list[str]) -> None:
    sql_query = """
    UPDATE public.repositories
        SET licenses = $1::varchar[],
        "updatedAt" = NOW()
    WHERE id = $2
      AND licenses IS DISTINCT FROM $1::varchar[]
    """
    await execute(sql_query, (licenses, repository_id))


async def mark_repo_as_processed(repo_id: str, repo_state: RepositoryState):
    sql_query = """
    UPDATE git."repositoryProcessing"
        SET "state" = $2,
        "lastProcessedAt" = NOW(),
        "updatedAt" = NOW(),
        "priority" = $3
    WHERE "repositoryId" = $1
    """
    result = await execute(sql_query, (repo_id, repo_state, RepositoryPriority.NORMAL))
    return str(result)


async def batch_insert_activities(records: list[tuple], batch_size=100):
    sql_query = """
    INSERT INTO integration.results(id, state, data, "tenantId", "integrationId")
    values($1, $2, $3, $4, $5)
    """
    logger.info(f"Saving {len(records)} activity into integration.results")
    for i in range(0, len(records), batch_size):
        batch = records[i : i + batch_size]
        await executemany(sql_query, batch)
    logger.info("activities saved into integration.results")


async def find_github_identity(github_username: str):
    sql_query = """
    SELECT id
        FROM "memberIdentities"
    WHERE
        platform = 'github'
        AND value = $1
        AND "verified" = TRUE
        AND "deletedAt" is null
    LIMIT 1
    """
    result = await fetchval(
        sql_query,
        (github_username,),
    )
    return result


async def find_maintainer_identity_by_email(email: str):
    sql_query = """
    SELECT id
        FROM "memberIdentities"
    WHERE platform IN ('github', 'git', 'gitlab')
        AND "verified" = TRUE
        AND value = $1
        AND "deletedAt" is null
    LIMIT 1
    """
    result = await fetchval(
        sql_query,
        (email,),
    )
    return result


async def upsert_maintainer(
    repo_id: str,
    identity_id: str,
    repo_url: str,
    role: str,
    original_role: str,
    start_date: datetime | None = None,
):
    sql_query = """
        INSERT INTO "maintainersInternal"
        ("role", "originalRole", "repoUrl", "repoId", "identityId", "startDate", "endDate")
        VALUES ($1, $2, $3, $4, $5, $6, $7)
        ON CONFLICT ("repoId", "identityId", role) DO UPDATE
        SET "originalRole" = EXCLUDED."originalRole",
            "repoUrl" = EXCLUDED."repoUrl",
            "startDate" = COALESCE("maintainersInternal"."startDate", EXCLUDED."startDate"),
            "endDate" = NULL,
            "updatedAt" = NOW()
    """
    await execute(
        sql_query,
        (role, original_role, repo_url, repo_id, identity_id, start_date, None),
    )


async def update_maintainer_run(repo_id: str, maintainer_file: str):
    """
    Update maintainer file info after processing.
    Updates git.repositoryProcessing table only.
    """
    sql_query = """
        UPDATE git."repositoryProcessing"
            SET "maintainerFile" = $1,
                "lastMaintainerRunAt" = NOW(),
                "updatedAt" = NOW()
        WHERE "repositoryId" = $2
    """
    await execute(
        sql_query,
        (maintainer_file, repo_id),
    )


async def get_maintainers_for_repo(repo_id: str):
    # Active rows only (endDate IS NULL) — reappearing maintainers hit the "new"
    # branch and get reactivated by upsert_maintainer's ON CONFLICT clause.
    # verified=TRUE mirrors find_github_identity / find_maintainer_identity_by_email.
    # platform/type are returned so the diff's safety guard can match identifiers
    # by kind and avoid cross-platform value collisions (e.g. a GitHub username
    # "foo" colliding with a same-named handle on another platform).
    maintainers_sql_query = """
        SELECT mi.role, mi."originalRole", mi."repoUrl", mi."repoId", mi."identityId",
               mem.value as identity_value, mem.platform, mem.type
            FROM "maintainersInternal" mi
            JOIN "memberIdentities" mem ON mi."identityId" = mem.id
        WHERE mi."repoId" = $1
          AND mi."endDate" IS NULL
          AND mem."verified" = TRUE
          AND mem."deletedAt" is null
        """
    return await query(
        maintainers_sql_query,
        (repo_id,),
    )


async def get_github_maintainer_usernames_for_repo(repo_id: str) -> set[str]:
    """Return GitHub usernames of active maintainers for fork/parent-repo filtering."""
    sql_query = """
        SELECT mem.value
            FROM "maintainersInternal" mi
            JOIN "memberIdentities" mem ON mi."identityId" = mem.id
        WHERE mi."repoId" = $1
          AND mi."endDate" IS NULL
          AND mem.platform = 'github'
          AND mem.type = 'username'
          AND mem."verified" = TRUE
          AND mem."deletedAt" is null
        """
    rows = await query(sql_query, (repo_id,))
    return {row["value"] for row in rows}


async def set_maintainer_end_date(
    repo_id: str, identity_id: str, role: str, change_date: datetime
):
    update_end_date_query = """
        UPDATE "maintainersInternal"
        SET "endDate" = $1,
            "updatedAt" = NOW()
        WHERE "repoId" = $2 AND "identityId" = $3 AND role = $4
    """
    await execute(
        update_end_date_query,
        (
            change_date,
            repo_id,
            identity_id,
            role,
        ),
    )


async def batch_check_parent_activities(
    activity_keys: list[tuple[str, str, str]],
    parent_channel: str,
    parent_segment_id: str,
) -> set[str]:
    """
    Batch check which activities exist in parent repo using full dedup key.

    Args:
        activity_keys: List of (timestamp, type, sourceId) tuples
        parent_channel: Parent repository URL
        parent_segment_id: Parent repository segment ID

    Returns:
        Set of sourceIds that exist in parent repo
    """
    if not activity_keys:
        return set()

    # Use dedup index with ALL fields for optimal performance
    # Index: (timestamp, platform, type, sourceId, channel, segmentId)
    # Build OR conditions for each (timestamp, type, sourceId) combination
    conditions = []
    params = ["git", parent_channel, parent_segment_id]
    param_idx = 4

    for timestamp_str, activity_type, source_id in activity_keys:
        conditions.append(
            f'("timestamp" = ${param_idx} AND "type" = ${param_idx + 1} AND "sourceId" = ${param_idx + 2})'
        )
        timestamp = datetime.fromisoformat(timestamp_str)
        params.append(timestamp)
        params.append(activity_type)
        params.append(source_id)
        param_idx += 3

    sql_query = f"""
    SELECT DISTINCT "sourceId"
    FROM "activityRelations"
    WHERE "platform" = $1
        AND "channel" = $2
        AND "segmentId" = $3
        AND ({" OR ".join(conditions)})
    """

    result = await query(sql_query, tuple(params))

    return {row["sourceId"] for row in result}


async def save_service_execution(service_execution: ServiceExecution) -> None:
    """
    Save service execution record to database.
    """
    try:
        sql_query = """
        INSERT INTO git."serviceExecutions" (
            "repoId", "operationType", "status", "errorCode",
            "errorMessage", "executionTimeSec", "metrics"
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7)
        """

        db_data = service_execution.to_db_dict()
        await execute(
            sql_query,
            (
                db_data["repoId"],
                db_data["operationType"],
                db_data["status"],
                db_data["errorCode"],
                db_data["errorMessage"],
                db_data["executionTimeSec"],
                db_data["metrics"],
            ),
        )
        logger.debug(
            f"Successfully saved service execution: {service_execution.operation_type} for repo {service_execution.repo_id}"
        )

    except Exception as e:
        logger.error(
            f"Failed to save service execution record: operation={service_execution.operation_type}, "
            f"repo_id={service_execution.repo_id}, status={service_execution.status.value}, "
            f"error: {e}"
        )
        # Do not re-raise - we don't want metrics saving to disrupt main operations


async def get_repo_affiliation_registry(repo_id: str) -> RepoAffiliationRegistry | None:
    sql_query = """
        SELECT "repoId", "filePath", "fileHash", "status", "snapshot", "lastRunAt"
        FROM git."repoAffiliationRegistry"
        WHERE "repoId" = $1
    """
    result = await fetchrow(sql_query, (repo_id,))
    if not result:
        return None

    return RepoAffiliationRegistry.from_db(dict(result))


async def upsert_repo_affiliation_registry(registry: RepoAffiliationRegistry) -> None:
    snapshot_json = registry.snapshot_for_db()
    sql_query = """
        INSERT INTO git."repoAffiliationRegistry" (
            "repoId", "filePath", "fileHash", "status", "snapshot", "lastRunAt", "updatedAt"
        )
        VALUES ($1, $2, $3, $4, $5::jsonb, NOW(), NOW())
        ON CONFLICT ("repoId") DO UPDATE SET
            "filePath" = EXCLUDED."filePath",
            "fileHash" = EXCLUDED."fileHash",
            "status" = EXCLUDED."status",
            "snapshot" = EXCLUDED."snapshot",
            "lastRunAt" = NOW(),
            "updatedAt" = NOW()
    """
    await execute(
        sql_query,
        (
            registry.repo_id,
            registry.file_path,
            registry.file_hash,
            registry.status,
            snapshot_json,
        ),
    )


async def find_many_member_ids_by_identities(identities: list[dict]) -> list[dict]:
    if not identities:
        return []

    values_parts: list[str] = []
    params: list[str | bool | int] = []
    param_index = 1
    for idx, identity in enumerate(identities):
        values_parts.append(
            f"(${param_index}::int, ${param_index + 1}::text, ${param_index + 2}::boolean,"
            f" ${param_index + 3}::text, ${param_index + 4}::text)"
        )
        params.extend(
            [
                idx,
                identity["type"],
                identity.get("verified", True),
                identity.get("platform"),
                identity["value"],
            ]
        )
        param_index += 5

    matches_by_idx: dict[int, set[str]] = {}
    rows = await query(
        f"""
        WITH input_identities (idx, identity_type, verified, platform, value) AS (
            VALUES {", ".join(values_parts)}
        )
        SELECT i.idx, mi."memberId"
        FROM input_identities i
        LEFT JOIN "memberIdentities" mi
            ON mi.type = i.identity_type
            AND mi.verified = i.verified
            AND lower(mi.value) = lower(i.value)
            AND mi.platform = i.platform
            AND mi."deletedAt" IS NULL
        ORDER BY i.idx
        """,
        tuple(params),
    )
    for row in rows:
        if row["memberId"] is None:
            continue
        matches_by_idx.setdefault(row["idx"], set()).add(str(row["memberId"]))

    results: list[dict] = []
    for idx, identity in enumerate(identities):
        member_ids = matches_by_idx.get(idx, set())
        member_id = next(iter(member_ids)) if len(member_ids) == 1 else None
        results.append(
            {
                "type": identity["type"],
                "platform": identity.get("platform"),
                "value": identity["value"],
                "verified": identity.get("verified", True),
                "member_id": member_id,
            }
        )

    return results


async def find_many_organization_ids_by_identities(identities: list[dict]) -> list[dict]:
    if not identities:
        return []

    values_parts: list[str] = []
    params: list[str | bool | int] = []
    param_index = 1
    for idx, identity in enumerate(identities):
        values_parts.append(
            f"(${param_index}::int, ${param_index + 1}::text, ${param_index + 2}::boolean,"
            f" ${param_index + 3}::text, ${param_index + 4}::text)"
        )
        params.extend(
            [
                idx,
                identity["type"],
                identity.get("verified", True),
                identity["platform"],
                identity["value"],
            ]
        )
        param_index += 5

    matches_by_idx: dict[int, set[str]] = {}
    rows = await query(
        f"""
        WITH input_identities (idx, identity_type, verified, platform, value) AS (
            VALUES {", ".join(values_parts)}
        )
        SELECT i.idx, oi."organizationId"
        FROM input_identities i
        LEFT JOIN "organizationIdentities" oi
            ON oi.type = i.identity_type
            AND oi.verified = i.verified
            AND oi.platform = i.platform
            AND lower(oi.value) = lower(i.value)
        ORDER BY i.idx
        """,
        tuple(params),
    )
    for row in rows:
        if row["organizationId"] is None:
            continue
        matches_by_idx.setdefault(row["idx"], set()).add(str(row["organizationId"]))

    results: list[dict] = []
    for idx, identity in enumerate(identities):
        organization_ids = matches_by_idx.get(idx, set())
        organization_id = next(iter(organization_ids)) if len(organization_ids) == 1 else None
        results.append(
            {
                "type": identity["type"],
                "platform": identity["platform"],
                "value": identity["value"],
                "verified": identity.get("verified", True),
                "organization_id": organization_id,
            }
        )

    return results


async def fetch_member_organizations(member_ids: list[str]) -> list[dict]:
    if not member_ids:
        return []

    return await query(
        """
        SELECT "memberId", "organizationId", "dateStart", "dateEnd", source, "deletedAt"
        FROM "memberOrganizations"
        WHERE "memberId" = ANY($1::uuid[])
        """,
        (member_ids,),
    )


async def fetch_segment_affiliations(member_ids: list[str], segment_id: str) -> list[dict]:
    """MSA rows are per segment — filter by segment_id so guards match this repo's project."""
    if not member_ids:
        return []

    return await query(
        """
        SELECT "memberId", "segmentId", "organizationId", "dateStart", "dateEnd", verified, "deletedAt"
        FROM "memberSegmentAffiliations"
        WHERE "memberId" = ANY($1::uuid[])
            AND "segmentId" = $2::uuid
            AND "organizationId" IS NOT NULL
        """,
        (member_ids, segment_id),
    )


async def insert_member_organizations(rows: list[dict]) -> None:
    if not rows:
        return

    undated_rows: list[tuple] = []
    open_ended_rows: list[tuple] = []
    dated_rows: list[tuple] = []

    for row in rows:
        params = (
            row["member_id"],
            row["organization_id"],
            row.get("date_start"),
            row.get("date_end"),
            row["source"],
        )
        date_start = row.get("date_start")
        date_end = row.get("date_end")
        if date_start is None and date_end is None:
            undated_rows.append(params)
        elif date_end is None:
            open_ended_rows.append(params)
        else:
            dated_rows.append(params)

    insert_sql = """
        INSERT INTO "memberOrganizations"(
            "memberId",
            "organizationId",
            "dateStart",
            "dateEnd",
            title,
            source,
            "createdAt",
            "updatedAt"
        )
        VALUES ($1, $2, $3, $4, NULL, $5, NOW(), NOW())
    """

    if undated_rows:
        sql = (
            insert_sql
            + """
            ON CONFLICT ("memberId", "organizationId")
                WHERE ("dateStart" IS NULL AND "dateEnd" IS NULL AND "deletedAt" IS NULL)
            DO NOTHING
        """
        )
        await executemany(sql, undated_rows)

    if open_ended_rows:
        sql = (
            insert_sql
            + """
            ON CONFLICT ("memberId", "organizationId", "dateStart")
                WHERE ("dateEnd" IS NULL AND "deletedAt" IS NULL)
            DO NOTHING
        """
        )
        await executemany(sql, open_ended_rows)

    if dated_rows:
        sql = (
            insert_sql
            + """
            ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd")
                WHERE ("deletedAt" IS NULL)
            DO NOTHING
        """
        )
        await executemany(sql, dated_rows)


async def insert_member_segment_affiliations(rows: list[dict]) -> None:
    if not rows:
        return

    await executemany(
        """
        INSERT INTO "memberSegmentAffiliations"(
            id,
            "memberId",
            "segmentId",
            "organizationId",
            "dateStart",
            "dateEnd"
        )
        VALUES (gen_random_uuid(), $1, $2, $3, $4, $5)
        """,
        [
            (
                row["member_id"],
                row["segment_id"],
                row["organization_id"],
                row.get("date_start"),
                row.get("date_end"),
            )
            for row in rows
        ],
    )