use super::{PgExecResult, PgOptResult, PgResult};
use crate::indexer::records::*;
use crate::utils::{blob_ref, strongref_to_parts};
use chrono::prelude::*;
use deadpool_postgres::GenericClient;
use ipld_core::cid::Cid;

pub async fn record_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO records (at_uri, did, cid) VALUES ($1, $2, $3) ON CONFLICT (at_uri) DO UPDATE SET cid=EXCLUDED.cid",
        &[&at_uri, &repo, &cid.to_bytes()],
    ).await
}

pub async fn record_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM records WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn block_insert<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
    rec: AppBskyGraphBlock,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO blocks (rkey, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
        &[&rkey, &repo, &rec.subject, &rec.created_at],
    ).await
}

pub async fn block_delete<C: GenericClient>(conn: &mut C, rkey: &str, repo: &str) -> PgExecResult {
    conn.execute(
        "DELETE FROM blocks WHERE rkey=$1 AND did=$2",
        &[&rkey, &repo],
    )
    .await
}

pub async fn chat_decl_upsert<C: GenericClient>(
    conn: &mut C,
    repo: &str,
    rec: ChatBskyActorDeclaration,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO chat_decls (did, allow_incoming) VALUES ($1, $2) ON CONFLICT (did) DO UPDATE SET allow_incoming=EXCLUDED.allow_incoming",
    &[&repo, &rec.allow_incoming.to_string()]
    ).await
}

pub async fn chat_decl_delete<C: GenericClient>(conn: &mut C, repo: &str) -> PgExecResult {
    conn.execute("DELETE FROM chat_decls WHERE did=$1", &[&repo])
        .await
}

pub async fn feedgen_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
    rec: AppBskyFeedGenerator,
) -> PgResult<bool> {
    let cid = cid.to_string();
    let description_facets = rec
        .description_facets
        .and_then(|v| serde_json::to_value(v).ok());
    let avatar = blob_ref(rec.avatar);

    conn.query_one(
        include_str!("sql/feedgen_upsert.sql"),
        &[
            &at_uri,
            &repo,
            &cid,
            &rec.did,
            &rec.content_mode,
            &rec.display_name,
            &rec.description,
            &description_facets,
            &avatar,
            &rec.created_at,
        ],
    )
    .await
    .map(|r| r.get::<_, i32>(0) == 0)
}

pub async fn feedgen_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM feedgens WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn follow_insert<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
    rec: AppBskyGraphFollow,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO follows (rkey, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
        &[&rkey, &repo, &rec.subject, &rec.created_at],
    ).await
}

pub async fn follow_delete<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
) -> PgOptResult<String> {
    let res = conn
        .query_opt(
            "DELETE FROM follows WHERE rkey=$1 AND did=$2 RETURNING subject",
            &[&rkey, &repo],
        )
        .await?;

    Ok(res.map(|v| v.get(0)))
}

pub async fn labeler_upsert<C: GenericClient>(
    conn: &mut C,
    repo: &str,
    cid: Cid,
    rec: AppBskyLabelerService,
) -> PgExecResult {
    let cid = cid.to_string();
    let reasons = rec
        .reason_types
        .as_ref()
        .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<_>>());
    let subject_types = rec
        .subject_types
        .as_ref()
        .map(|v| v.iter().map(|v| v.to_string()).collect::<Vec<_>>());

    conn.execute(
        include_str!("sql/label_service_upsert.sql"),
        &[
            &repo,
            &cid,
            &reasons,
            &subject_types,
            &rec.subject_collections,
        ],
    )
    .await?;

    super::maintain_label_defs(conn, repo, &rec).await
}

pub async fn labeler_delete<C: GenericClient>(conn: &mut C, repo: &str) -> PgExecResult {
    conn.execute("DELETE FROM labelers WHERE did=$1", &[&repo])
        .await
}

pub async fn like_insert<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
    rec: AppBskyFeedLike,
) -> PgExecResult {
    let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref());

    conn.execute(
        "INSERT INTO likes (rkey, did, subject, subject_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)",
        &[&rkey, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), &via_uri, &via_cid, &rec.created_at]
    ).await
}

pub async fn like_delete<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
) -> PgOptResult<String> {
    let res = conn
        .query_opt(
            "DELETE FROM likes WHERE rkey=$1 AND did=$2 RETURNING subject",
            &[&rkey, &repo],
        )
        .await?;

    Ok(res.map(|v| v.get(0)))
}

pub async fn list_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
    rec: AppBskyGraphList,
) -> PgResult<bool> {
    let cid = cid.to_string();
    let description_facets = rec
        .description_facets
        .and_then(|v| serde_json::to_value(v).ok());
    let avatar = blob_ref(rec.avatar);

    conn.query_one(
        include_str!("sql/list_upsert.sql"),
        &[
            &at_uri,
            &repo,
            &cid,
            &rec.purpose,
            &rec.name,
            &rec.description,
            &description_facets,
            &avatar,
            &rec.created_at,
        ],
    )
    .await
    .map(|r| r.get::<_, i32>(0) == 0)
}

pub async fn list_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM lists WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn list_block_insert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    rec: AppBskyGraphListBlock,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO list_blocks (at_uri, did, list_uri, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
        &[&at_uri, &repo, &rec.subject, &rec.created_at],
    ).await
}

pub async fn list_block_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM list_blocks WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn list_item_insert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    rec: AppBskyGraphListItem,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO list_items (at_uri, list_uri, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING",
        &[&at_uri, &rec.list, &rec.subject, &rec.created_at],
    ).await
}

pub async fn list_item_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM list_items WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn notif_decl_upsert<C: GenericClient>(
    conn: &mut C,
    repo: &str,
    rec: AppBskyNotificationDeclaration,
) -> PgExecResult {
    conn.execute(
        "INSERT INTO notif_decl (did, allow_subscriptions) VALUES ($1, $2) ON CONFLICT (did) DO UPDATE SET allow_subscriptions=EXCLUDED.allow_subscriptions",
        &[&repo, &rec.allow_subscriptions.to_string()],
    ).await
}

pub async fn notif_decl_delete<C: GenericClient>(conn: &mut C, repo: &str) -> PgExecResult {
    conn.execute("DELETE FROM notif_decl WHERE did=$1", &[&repo])
        .await
}

pub async fn post_insert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
    rec: AppBskyFeedPost,
) -> PgExecResult {
    let cid = cid.to_string();
    let record = serde_json::to_value(&rec).unwrap();
    let facets = rec.facets.and_then(|v| serde_json::to_value(v).ok());
    let (parent_uri, parent_cid) = strongref_to_parts(rec.reply.as_ref().map(|v| &v.parent));
    let (root_uri, root_cid) = strongref_to_parts(rec.reply.as_ref().map(|v| &v.root));
    let embed = rec.embed.as_ref().map(|v| v.as_str());
    let embed_subtype = rec.embed.as_ref().and_then(|v| v.subtype());

    let count = conn
        .execute(
            include_str!("sql/post_insert.sql"),
            &[
                &at_uri,
                &repo,
                &cid,
                &record,
                &rec.text,
                &facets,
                &rec.langs.unwrap_or_default(),
                &rec.tags.unwrap_or_default(),
                &parent_uri,
                &parent_cid,
                &root_uri,
                &root_cid,
                &embed,
                &embed_subtype,
                &rec.created_at,
            ],
        )
        .await?;

    if let Some(embed) = rec.embed.and_then(|embed| embed.into_bsky()) {
        post_embed_insert(conn, at_uri, embed, rec.created_at).await?;
    }

    Ok(count)
}

pub async fn post_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM posts WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn post_get_info_for_delete<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
) -> PgOptResult<(Option<String>, Option<String>)> {
    let res = conn
        .query_opt(
            "SELECT parent_uri, per.uri FROM posts LEFT JOIN post_embed_record per on at_uri = per.post_uri WHERE at_uri = $1",
            &[&at_uri],
        )
        .await?;

    Ok(res.map(|row| (row.get(0), row.get(1))))
}

pub async fn post_embed_insert<C: GenericClient>(
    conn: &mut C,
    post: &str,
    embed: AppBskyEmbed,
    created_at: DateTime<Utc>,
) -> PgExecResult {
    match embed {
        AppBskyEmbed::Images(embed) => post_embed_image_insert(conn, post, embed).await,
        AppBskyEmbed::Video(embed) => post_embed_video_insert(conn, post, embed).await,
        AppBskyEmbed::External(embed) => post_embed_external_insert(conn, post, embed).await,
        AppBskyEmbed::Record(embed) => {
            post_embed_record_insert(conn, post, embed, created_at).await
        }
        AppBskyEmbed::RecordWithMedia(embed) => {
            post_embed_record_insert(conn, post, embed.record, created_at).await?;
            match *embed.media {
                AppBskyEmbed::Images(embed) => post_embed_image_insert(conn, post, embed).await,
                AppBskyEmbed::Video(embed) => post_embed_video_insert(conn, post, embed).await,
                AppBskyEmbed::External(embed) => {
                    post_embed_external_insert(conn, post, embed).await
                }
                _ => unreachable!(),
            }
        }
    }
}

async fn post_embed_image_insert<C: GenericClient>(
    conn: &mut C,
    post: &str,
    embed: AppBskyEmbedImages,
) -> PgExecResult {
    let stmt = conn.prepare("INSERT INTO post_embed_images (post_uri, seq, cid, mime_type, alt, width, height) VALUES ($1, $2, $3, $4, $5, $6, $7)").await?;

    for (idx, image) in embed.images.iter().enumerate() {
        let cid = image.image.cid.to_string();
        let width = image.aspect_ratio.as_ref().map(|v| v.width);
        let height = image.aspect_ratio.as_ref().map(|v| v.height);

        conn.execute(
            &stmt,
            &[
                &post,
                &(idx as i16),
                &cid,
                &image.image.mime_type,
                &image.alt,
                &width,
                &height,
            ],
        )
        .await?;
    }

    Ok(0)
}

async fn post_embed_video_insert<C: GenericClient>(
    conn: &mut C,
    post: &str,
    embed: AppBskyEmbedVideo,
) -> PgExecResult {
    let cid = embed.video.cid.to_string();
    let width = embed.aspect_ratio.as_ref().map(|v| v.width);
    let height = embed.aspect_ratio.as_ref().map(|v| v.height);

    let count = conn.execute(
        "INSERT INTO post_embed_video (post_uri, cid, mime_type, alt, width, height) VALUES ($1, $2, $3, $4, $5, $6)",
        &[&post, &cid, &embed.video.mime_type, &embed.alt, &width, &height],
    ).await?;

    if let Some(captions) = embed.captions {
        let stmt = conn.prepare_cached("INSERT INTO post_embed_video_captions (post_uri, cid, mime_type, language) VALUES ($1, $2, $3, $4)").await?;

        for caption in captions {
            let cid = caption.file.cid.to_string();
            conn.execute(
                &stmt,
                &[&post, &cid, &caption.file.mime_type, &caption.lang],
            )
            .await?;
        }
    }

    Ok(count)
}

async fn post_embed_external_insert<C: GenericClient>(
    conn: &mut C,
    post: &str,
    embed: AppBskyEmbedExternal,
) -> PgExecResult {
    let thumb_mime = embed.external.thumb.as_ref().map(|v| v.mime_type.clone());
    let thumb_cid = embed.external.thumb.as_ref().map(|v| v.cid.to_string());

    conn.execute(
        "INSERT INTO post_embed_ext (post_uri, uri, title, description, thumb_mime_type, thumb_cid) VALUES ($1, $2, $3, $4, $5, $6)",
        &[&post, &embed.external.uri, &embed.external.title, &embed.external.description, &thumb_mime, &thumb_cid],
    ).await
}

async fn post_embed_record_insert<C: GenericClient>(
    conn: &mut C,
    post: &str,
    embed: AppBskyEmbedRecord,
    post_created_at: DateTime<Utc>,
) -> PgExecResult {
    // strip "at://" then break into parts by '/'
    let parts = embed.record.uri[5..].split('/').collect::<Vec<_>>();

    let detached = if parts[1] == "app.bsky.feed.post" {
        let postgate_effective: Option<DateTime<Utc>> = conn
            .query_opt(
                "SELECT created_at FROM postgates WHERE post_uri=$1",
                &[&post],
            )
            .await?
            .map(|v| v.get(0));

        postgate_effective
            .map(|v| Utc::now().min(post_created_at) > v)
            .unwrap_or_default()
    } else {
        false
    };

    conn.execute(
        "INSERT INTO post_embed_record (post_uri, record_type, uri, cid, detached) VALUES ($1, $2, $3, $4, $5)",
        &[&post, &parts[1], &embed.record.uri, &embed.record.cid.to_string(), &detached],
    ).await
}

pub async fn postgate_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    cid: Cid,
    rec: &AppBskyFeedPostgate,
) -> PgExecResult {
    let rules = rec
        .embedding_rules
        .iter()
        .map(|v| v.as_str().to_string())
        .collect::<Vec<_>>();

    conn.execute(
        include_str!("sql/postgate_upsert.sql"),
        &[
            &at_uri,
            &cid.to_string(),
            &rec.post,
            &rec.detached_embedding_uris,
            &rules,
            &rec.created_at,
        ],
    )
    .await
}

pub async fn postgate_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM postgates WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn postgate_maintain_detaches<C: GenericClient>(
    conn: &mut C,
    post: &str,
    detached: &[String],
    disable_effective: Option<NaiveDateTime>,
) -> PgExecResult {
    conn.execute(
        "SELECT maintain_postgates($1, $2, $3)",
        &[&post, &detached, &disable_effective],
    )
    .await
}

pub async fn profile_upsert<C: GenericClient>(
    conn: &mut C,
    repo: &str,
    cid: Cid,
    rec: AppBskyActorProfile,
) -> PgExecResult {
    let cid = cid.to_string();
    let avatar = blob_ref(rec.avatar);
    let banner = blob_ref(rec.banner);
    let (pinned_uri, pinned_cid) = strongref_to_parts(rec.pinned_post.as_ref());
    let (joined_sp_uri, joined_sp_cid) = strongref_to_parts(rec.joined_via_starter_pack.as_ref());

    conn.execute(
        include_str!("sql/profile_upsert.sql"),
        &[
            &repo,
            &cid,
            &avatar,
            &banner,
            &rec.display_name,
            &rec.description,
            &pinned_uri,
            &pinned_cid,
            &joined_sp_uri,
            &joined_sp_cid,
            &rec.created_at.unwrap_or(Utc::now()).naive_utc(),
        ],
    )
    .await
}

pub async fn profile_delete<C: GenericClient>(conn: &mut C, repo: &str) -> PgExecResult {
    conn.execute("DELETE FROM profiles WHERE did=$1", &[&repo])
        .await
}

pub async fn repost_insert<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
    rec: AppBskyFeedRepost,
) -> PgExecResult {
    let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref());

    conn.execute(
        "INSERT INTO reposts (rkey, did, post, post_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)",
        &[
            &rkey,
            &repo,
            &rec.subject.uri,
            &rec.subject.cid.to_string(),
            &via_uri,
            &via_cid,
            &rec.created_at,
        ],
    )
    .await
}

pub async fn repost_delete<C: GenericClient>(
    conn: &mut C,
    rkey: &str,
    repo: &str,
) -> PgOptResult<String> {
    let res = conn
        .query_opt(
            "DELETE FROM reposts WHERE rkey=$1 AND did=$2 RETURNING post",
            &[&rkey, &repo],
        )
        .await?;

    Ok(res.map(|v| v.get(0)))
}

pub async fn starter_pack_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
    rec: AppBskyGraphStarterPack,
) -> PgResult<bool> {
    let cid = cid.to_string();
    let record = serde_json::to_value(&rec).unwrap();
    let description_facets = rec
        .description_facets
        .and_then(|v| serde_json::to_value(v).ok());
    let feeds = rec
        .feeds
        .map(|v| v.into_iter().map(|item| item.uri).collect::<Vec<_>>());

    conn.query_one(
        include_str!("sql/starterpack_upsert.sql"),
        &[
            &at_uri,
            &repo,
            &cid,
            &record,
            &rec.name,
            &rec.description,
            &description_facets,
            &rec.list,
            &feeds,
            &rec.created_at,
        ],
    )
    .await
    .map(|r| r.get::<_, i32>(0) == 0)
}

pub async fn starter_pack_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM starterpacks WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn status_upsert<C: GenericClient>(
    conn: &mut C,
    repo: &str,
    rec: AppBskyActorStatus,
) -> PgExecResult {
    let record = serde_json::to_value(&rec).unwrap();
    let thumb = rec.embed.as_ref().and_then(|v| v.external.thumb.clone());
    let thumb_mime = thumb.as_ref().map(|v| v.mime_type.clone());
    let thumb_cid = thumb.as_ref().map(|v| v.cid.to_string());

    conn.execute(
        include_str!("sql/status_upsert.sql"),
        &[
            &repo,
            &rec.status.to_string(),
            &rec.duration_minutes,
            &record,
            &rec.embed.as_ref().map(|v| v.external.uri.clone()),
            &rec.embed.as_ref().map(|v| v.external.title.clone()),
            &rec.embed.as_ref().map(|v| v.external.description.clone()),
            &thumb_mime,
            &thumb_cid,
            &rec.created_at,
        ],
    )
    .await
}

pub async fn status_delete<C: GenericClient>(conn: &mut C, did: &str) -> PgExecResult {
    conn.execute("DELETE FROM statuses WHERE did=$1", &[&did])
        .await
}

pub async fn threadgate_upsert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    cid: Cid,
    rec: AppBskyFeedThreadgate,
) -> PgExecResult {
    let record = serde_json::to_value(&rec).unwrap();

    let allowed_lists = rec
        .allow
        .iter()
        .filter_map(|rule| match rule {
            ThreadgateRule::List { list } => Some(list.clone()),
            _ => None,
        })
        .collect::<Vec<_>>();

    let allow = rec
        .allow
        .into_iter()
        .map(|v| v.as_str().to_string())
        .collect::<Vec<_>>();

    conn.execute(
        include_str!("sql/threadgate_upsert.sql"),
        &[
            &at_uri,
            &cid.to_string(),
            &rec.post,
            &rec.hidden_replies,
            &allow,
            &allowed_lists,
            &record,
            &rec.created_at,
        ],
    )
    .await
}

pub async fn threadgate_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM threadgates WHERE at_uri=$1", &[&at_uri])
        .await
}

pub async fn verification_insert<C: GenericClient>(
    conn: &mut C,
    at_uri: &str,
    repo: &str,
    cid: Cid,
    rec: AppBskyGraphVerification,
) -> PgExecResult {
    let cid = cid.to_string();

    conn.execute(
        "INSERT INTO verification (at_uri, verifier, cid, subject, handle, display_name, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING",
        &[&at_uri, &repo, &cid, &rec.subject, &rec.handle, &rec.display_name, &rec.created_at],
    ).await
}

pub async fn verification_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
    conn.execute("DELETE FROM verification WHERE at_uri=$1", &[&at_uri])
        .await
}