#![deny(warnings, rust_2018_idioms)]

use anyhow::{bail, Context, Result};
use std::convert::TryFrom;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::vec;

use bytes::{Buf, BufMut, BytesMut};
use packed_struct::prelude::*;

use crate::codec::{character_string, domain_name, message};
use crate::fbs::dns_enums_conv;
use crate::fbs::dns_enums_generated::ResourceType;
use crate::fbs::dns_message_generated::{
    rdata, OPTArgs, OPTOption, OPTOptionArgs, ResourceData, ResourceDataArgs, OPT,
};

pub fn write_rdata(
    resource_type: u16,
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    if let Some(raw_data) = rdata.raw_data() {
        // Pass through the raw data directly.
        // In this case, we also don't require an explicitly supported resource type enum.
        buf.extend_from_slice(&raw_data);
        return Ok(());
    }

    return match dns_enums_conv::resourcetype_int(resource_type as usize) {
        Some(ResourceType::TYPE_A) => write_a(rdata, buf),
        Some(ResourceType::TYPE_NS) => write_ns(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_CNAME) => write_cname(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_SOA) => write_soa(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_PTR) => write_ptr(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_HINFO) => write_hinfo(rdata, buf),
        Some(ResourceType::TYPE_MX) => write_mx(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_TXT) => write_txt(rdata, buf),
        Some(ResourceType::TYPE_RP) => write_rp(rdata, buf, ptr_offsets),
        Some(ResourceType::TYPE_AFSDB) => write_afsdb(rdata, buf, ptr_offsets),
        //Some(ResourceType::TYPE_SIG) => bail!("TODO(#1) writer for SIG resources"),
        //Some(ResourceType::TYPE_KEY) => bail!("TODO(#1) writer for KEY resources"),
        Some(ResourceType::TYPE_AAAA) => write_aaaa(rdata, buf),
        //Some(ResourceType::TYPE_LOC) => bail!("TODO(#1) writer for LOC resources"),
        Some(ResourceType::TYPE_SRV) => write_srv(rdata, buf),
        //Some(ResourceType::TYPE_NAPTR) => bail!("TODO(#1) writer for NAPTR resources"),
        //Some(ResourceType::TYPE_KX) => bail!("TODO(#1) writer for KX resources"),
        //Some(ResourceType::TYPE_CERT) => bail!("TODO(#1) writer for CERT resources"),
        Some(ResourceType::TYPE_DNAME) => write_dname(rdata, buf),
        Some(ResourceType::TYPE_OPT) => {
            bail!(
                "Writing resource.type={} must be done using rdata::write_opt",
                resource_type
            );
        }
        //Some(ResourceType::TYPE_APL) => bail!("TODO(#1) writer for APL resources"),
        Some(ResourceType::TYPE_DS) => write_ds(rdata, buf),
        //Some(ResourceType::TYPE_SSHFP) => bail!("TODO(#1) writer for SSHFP resources"),
        //Some(ResourceType::TYPE_IPSECKEY) => bail!("TODO(#1) writer for IPSECKEY resources"),
        Some(ResourceType::TYPE_RRSIG) => write_rrsig(rdata, buf),
        Some(ResourceType::TYPE_NSEC) => write_nsec(rdata, buf),
        Some(ResourceType::TYPE_DNSKEY) => write_dnskey(rdata, buf),
        //Some(ResourceType::TYPE_DHCID) => bail!("TODO(#1) writer for DHCID resources"),
        //Some(ResourceType::TYPE_NSEC3) => bail!("TODO(#1) writer for NSEC3 resources"),
        //Some(ResourceType::TYPE_NSEC3PARAM) => bail!("TODO(#1) writer for NSEC3PARAM resources"),
        //Some(ResourceType::TYPE_TLSA) => bail!("TODO(#1) writer for TLSA resources"),
        //Some(ResourceType::TYPE_SMIMEA) => bail!("TODO(#1) writer for SMIMEA resources"),
        //Some(ResourceType::TYPE_HIP) => bail!("TODO(#1) writer for HIP resources"),
        //Some(ResourceType::TYPE_OPENPGPKEY) => bail!("TODO(#1) writer for OPENPGPKEY resources"),
        //Some(ResourceType::TYPE_CSYNC) => bail!("TODO(#1) writer for CSYNC resources"),
        //Some(ResourceType::TYPE_TKEY) => bail!("TODO(#1) writer for TKEY resources"),
        //Some(ResourceType::TYPE_TSIG) => bail!("TODO(#1) writer for TSIG resources"),
        //Some(ResourceType::TYPE_URI) => bail!("TODO(#1) writer for URI resources"),
        //Some(ResourceType::TYPE_CAA) => bail!("TODO(#1) writer for CAA resources"),
        // We don't have an explicit encoder for this.
        // Should've provided rdata.raw_data, or should have left rdata unset in the resource entirely.
        _ => bail!("Reading resource.type={} is unsupported", resource_type),
    };
}

pub fn read_rdata<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    resource_type: u16,
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<ResourceDataArgs<'a>> {
    let mut args = ResourceDataArgs::default();
    match dns_enums_conv::resourcetype_int(resource_type as usize) {
        Some(ResourceType::TYPE_A) => {
            args.a = Some(read_a(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_NS) => {
            args.ns = Some(read_ns(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_CNAME) => {
            args.cname = Some(read_cname(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_SOA) => {
            args.soa = Some(read_soa(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_PTR) => {
            args.ptr = Some(read_ptr(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_HINFO) => {
            args.hinfo = Some(read_hinfo(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_MX) => {
            args.mx = Some(read_mx(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_TXT) => {
            args.txt = Some(read_txt(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_RP) => {
            args.rp = Some(read_rp(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_AFSDB) => {
            args.afsdb = Some(read_afsdb(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        //Some(ResourceType::TYPE_SIG) => bail!("TODO(#1) reader for SIG resources"),
        //Some(ResourceType::TYPE_KEY) => bail!("TODO(#1) reader for KEY resources"),
        Some(ResourceType::TYPE_AAAA) => {
            args.aaaa = Some(read_aaaa(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        //Some(ResourceType::TYPE_LOC) => bail!("TODO(#1) reader for LOC resources"),
        Some(ResourceType::TYPE_SRV) => {
            args.srv = Some(read_srv(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        //Some(ResourceType::TYPE_NAPTR) => bail!("TODO(#1) reader for NAPTR resources"),
        //Some(ResourceType::TYPE_KX) => bail!("TODO(#1) reader for KX resources"),
        //Some(ResourceType::TYPE_CERT) => bail!("TODO(#1) reader for CERT resources"),
        Some(ResourceType::TYPE_DNAME) => {
            args.dname = Some(read_dname(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_OPT) => {
            bail!(
                "Reading resource.type={} must be done using rdata::read_opt",
                resource_type
            );
        }
        //Some(ResourceType::TYPE_APL) => bail!("TODO(#1) reader for APL resources"),
        Some(ResourceType::TYPE_DS) => {
            args.ds = Some(read_ds(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        //Some(ResourceType::TYPE_SSHFP) => bail!("TODO(#1) reader for SSHFP resources"),
        //Some(ResourceType::TYPE_IPSECKEY) => bail!("TODO(#1) reader for IPSECKEY resources"),
        Some(ResourceType::TYPE_RRSIG) => {
            args.rrsig = Some(read_rrsig(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_NSEC) => {
            args.nsec = Some(read_nsec(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        Some(ResourceType::TYPE_DNSKEY) => {
            args.dnskey = Some(read_dnskey(fb_builder, buf, rdata_offset, rdata_len)?);
        }
        //Some(ResourceType::TYPE_DHCID) => bail!("TODO(#1) reader for DHCID resources"),
        //Some(ResourceType::TYPE_NSEC3) => bail!("TODO(#1) reader for NSEC3 resources"),
        //Some(ResourceType::TYPE_NSEC3PARAM) => bail!("TODO(#1) reader for NSEC3PARAM resources"),
        //Some(ResourceType::TYPE_TLSA) => bail!("TODO(#1) reader for TLSA resources"),
        //Some(ResourceType::TYPE_SMIMEA) => bail!("TODO(#1) reader for SMIMEA resources"),
        //Some(ResourceType::TYPE_HIP) => bail!("TODO(#1) reader for HIP resources"),
        //Some(ResourceType::TYPE_OPENPGPKEY) => bail!("TODO(#1) reader for OPENPGPKEY resources"),
        //Some(ResourceType::TYPE_CSYNC) => bail!("TODO(#1) reader for CSYNC resources"),
        //Some(ResourceType::TYPE_TKEY) => bail!("TODO(#1) reader for TKEY resources"),
        //Some(ResourceType::TYPE_TSIG) => bail!("TODO(#1) reader for TSIG resources"),
        //Some(ResourceType::TYPE_URI) => bail!("TODO(#1) reader for URI resources"),
        //Some(ResourceType::TYPE_CAA) => bail!("TODO(#1) reader for CAA resources"),
        // We don't have an explicit decoder for this, or we don't have an enum for it at all (not defined by IANA??).
        // Copy the data directly into rdata.raw_data.
        _ => {
            args.raw_data =
                Some(fb_builder.create_vector(&buf[rdata_offset..rdata_offset + rdata_len]));
        }
    }
    Ok(args)
}

// A, from RFC1035 section 3.3.13
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    ADDRESS                    |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+

#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataABits {
    #[packed_field(bits = "0:7")]
    address1: u8,
    #[packed_field(bits = "8:15")]
    address2: u8,
    #[packed_field(bits = "16:23")]
    address3: u8,
    #[packed_field(bits = "24:31")]
    address4: u8,
}

pub fn write_a_ip(ip: &Ipv4Addr, buf: &mut BytesMut) -> Result<()> {
    let octets = ip.octets();
    let packed_bits = RDataABits {
        address1: octets[0],
        address2: octets[1],
        address3: octets[2],
        address4: octets[3],
    }
    .pack();
    buf.extend_from_slice(&packed_bits);
    Ok(())
}

fn write_a(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let a = rdata.a().context("missing rdata.a")?;
    let packed_bits = RDataABits {
        address1: a.address1(),
        address2: a.address2(),
        address3: a.address3(),
        address4: a.address4(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);

    Ok(())
}

fn read_a<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::A<'a>>> {
    let bits_size = RDataABits::packed_bytes();
    if rdata_len != bits_size {
        bail!(
            "A record rdata requires {} bytes, got {}",
            bits_size,
            rdata_len
        );
    }
    let bits = RDataABits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack A record rdata bits")?;
    Ok(rdata::A::create(
        fb_builder,
        &rdata::AArgs {
            address1: bits.address1,
            address2: bits.address2,
            address3: bits.address3,
            address4: bits.address4,
        },
    ))
}

// NS, from RFC1035 section 3.3.11
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                   NSDNAME                     /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_ns(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    domain_name::write(
        rdata
            .ns()
            .context("missing rdata.ns")?
            .nsdname()
            .context("missing rdata.ns.nsdname")?,
        buf,
        ptr_offsets,
        "rdata.ns.nsdname",
    )?;
    Ok(())
}

fn read_ns<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::NS<'a>>> {
    let (name_bytes_consumed, name_str) = domain_name::read(buf, rdata_offset, "rdata.ns.nsdname")?;

    if name_bytes_consumed != rdata_len {
        bail!(
            "Expected NS record data to be {} bytes, but was {} bytes",
            rdata_len,
            name_bytes_consumed
        );
    }

    let args = rdata::NSArgs {
        nsdname: Some(fb_builder.create_string(name_str.as_str())),
    };
    Ok(rdata::NS::create(fb_builder, &args))
}

// CNAME, from RFC1035 section 3.3.1
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                     CNAME                     /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_cname(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    domain_name::write(
        rdata
            .cname()
            .context("missing rdata.cname")?
            .cname()
            .context("missing rdata.cname.cname")?,
        buf,
        ptr_offsets,
        "rdata.cname.cname",
    )?;
    Ok(())
}

fn read_cname<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::CNAME<'a>>> {
    let (name_bytes_consumed, name_str) =
        domain_name::read(buf, rdata_offset, "rdata.cname.cname")?;

    if name_bytes_consumed != rdata_len {
        bail!(
            "Expected CNAME record data to be {} bytes, but was {} bytes",
            rdata_len,
            name_bytes_consumed
        );
    }

    let args = rdata::CNAMEArgs {
        cname: Some(fb_builder.create_string(name_str.as_str())),
    };
    Ok(rdata::CNAME::create(fb_builder, &args))
}

// SOA, from RFC1035 section 3.3.13
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                     MNAME                     /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                     RNAME                     /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    SERIAL                     |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    REFRESH                    |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                     RETRY                     |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    EXPIRE                     |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    MINIMUM                    |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataSOABits {
    // mname: variable length domain string
    // rname: variable length domain string
    #[packed_field(bits = "0:31")]
    serial: u32,
    #[packed_field(bits = "32:63")]
    refresh: u32,
    #[packed_field(bits = "64:95")]
    retry: u32,
    #[packed_field(bits = "96:127")]
    expire: u32,
    #[packed_field(bits = "128:159")]
    minimum: u32,
}

fn write_soa(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    let soa = rdata.soa().context("missing rdata.soa")?;
    write_soa_fields(
        &SOAFields {
            mname: soa.mname().context("missing rdata.soa.mname")?,
            rname: soa.rname().context("missing rdata.soa.rname")?,
            serial: soa.serial(),
            refresh: soa.refresh(),
            retry: soa.retry(),
            expire: soa.expire(),
            minimum: soa.minimum(),
        },
        buf,
        ptr_offsets,
    )
}

pub struct SOAFields<'a> {
    pub mname: &'a str,
    pub rname: &'a str,
    pub serial: u32,
    pub refresh: u32,
    pub retry: u32,
    pub expire: u32,
    pub minimum: u32,
}

pub fn write_soa_fields<'a>(
    soa: &SOAFields<'a>,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    domain_name::write(soa.mname, buf, ptr_offsets, "rdata.soa.mname")?;
    domain_name::write(soa.rname, buf, ptr_offsets, "rdata.soa.rname")?;

    let packed_bits = RDataSOABits {
        serial: soa.serial,
        refresh: soa.refresh,
        retry: soa.retry,
        expire: soa.expire,
        minimum: soa.minimum,
    }
    .pack();
    buf.extend_from_slice(&packed_bits);

    Ok(())
}

fn read_soa<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::SOA<'a>>> {
    let bits_size = RDataSOABits::packed_bytes();
    if rdata_len < bits_size + 2 {
        bail!(
            "SOA record rdata requires at least {} bytes, got {}",
            bits_size + 2,
            rdata_len
        );
    }

    let mut offset = rdata_offset;

    let (mname_bytes_consumed, mname_str) = domain_name::read(buf, offset, "rdata.soa.mname")?;
    offset += mname_bytes_consumed;

    let (rname_bytes_consumed, rname_str) = domain_name::read(buf, offset, "rdata.soa.rname")?;
    offset += rname_bytes_consumed;

    if bits_size + (offset - rdata_offset) != rdata_len {
        bail!(
            "Expected SOA record data to be {} bytes, but was {} + ({} - {}) bytes",
            rdata_len,
            bits_size,
            offset,
            rdata_offset,
        );
    }
    let bits = RDataSOABits::unpack_from_slice(&buf[offset..offset + bits_size])
        .context("couldn't unpack SOA record rdata bits")?;

    let args = rdata::SOAArgs {
        mname: Some(fb_builder.create_string(mname_str.as_str())),
        rname: Some(fb_builder.create_string(rname_str.as_str())),
        serial: bits.serial,
        refresh: bits.refresh,
        retry: bits.retry,
        expire: bits.expire,
        minimum: bits.minimum,
    };
    Ok(rdata::SOA::create(fb_builder, &args))
}

// PTR, from RFC1035 section 3.3.12
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                   PTRDNAME                    /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_ptr(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    domain_name::write(
        rdata
            .ptr()
            .context("missing rdata.ptr")?
            .ptrdname()
            .context("missing rdata.ptr.ptrdname")?,
        buf,
        ptr_offsets,
        "rdata.ptr.ptrdname",
    )?;
    Ok(())
}

fn read_ptr<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::PTR<'a>>> {
    let (name_bytes_consumed, name_str) =
        domain_name::read(buf, rdata_offset, "rdata.ptr.ptrdname")?;

    if name_bytes_consumed != rdata_len {
        bail!(
            "Expected PTR record data to be {} bytes, but was {} bytes",
            rdata_len,
            name_bytes_consumed
        );
    }

    let args = rdata::PTRArgs {
        ptrdname: Some(fb_builder.create_string(name_str.as_str())),
    };
    Ok(rdata::PTR::create(fb_builder, &args))
}

// HINFO, from RFC1035 section 3.3.2
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                      CPU                      /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                       OS                      /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_hinfo(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let hinfo = rdata.hinfo().context("missing rdata.hinfo")?;
    character_string::write(
        hinfo.cpu().context("missing rdata.hinfo.cpu")?,
        buf,
        "rdata.hinfo.cpu",
    )?;
    character_string::write(
        hinfo.os().context("missing rdata.hinfo.os")?,
        buf,
        "rdata.hinfo.os",
    )?;
    Ok(())
}

fn read_hinfo<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::HINFO<'a>>> {
    let (cpu_bytes_consumed, cpu) = character_string::read(&buf, rdata_offset, "rdata.hinfo.cpu")?;
    let (os_bytes_consumed, os) =
        character_string::read(&buf, rdata_offset + cpu_bytes_consumed, "rdata.hinfo.os")?;

    if cpu_bytes_consumed + os_bytes_consumed != rdata_len {
        bail!(
            "Expected HINFO record data to be {} bytes, but was {} + {} bytes",
            rdata_len,
            cpu_bytes_consumed,
            os_bytes_consumed
        );
    }

    let args = rdata::HINFOArgs {
        cpu: Some(fb_builder.create_vector(cpu.bytes())),
        os: Some(fb_builder.create_vector(os.bytes())),
    };
    Ok(rdata::HINFO::create(fb_builder, &args))
}

// MX, from RFC1035 section 3.3.9
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                  PREFERENCE                   |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                   EXCHANGE                    /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataMXBits {
    #[packed_field(bits = "0:15")]
    preference: u16,
    // exchange: variable length domain string
}

fn write_mx(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    let mx = rdata.mx().context("missing rdata.mx")?;
    let packed_bits = RDataMXBits {
        preference: mx.preference(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);

    domain_name::write(
        mx.exchange().context("missing rdata.mx.exchange")?,
        buf,
        ptr_offsets,
        "rdata.mx.exchange",
    )?;

    Ok(())
}

fn read_mx<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    mut rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::MX<'a>>> {
    let bits_size = RDataMXBits::packed_bytes();
    if rdata_len < bits_size + 1 {
        bail!(
            "MX record rdata requires at least {} bytes, got {}",
            bits_size + 1,
            rdata_len
        );
    }
    let bits = RDataMXBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack MX record rdata bits")?;
    rdata_offset += bits_size;

    let (exchange_bytes_consumed, exchange_str) =
        domain_name::read(buf, rdata_offset, "rdata.mx.exchange")?;
    if bits_size + exchange_bytes_consumed != rdata_len {
        bail!(
            "Expected MX record data to be {} bytes, but was {} + {} bytes",
            rdata_len,
            bits_size,
            exchange_bytes_consumed
        );
    }

    let args = rdata::MXArgs {
        preference: bits.preference,
        exchange: Some(fb_builder.create_string(exchange_str.as_str())),
    };
    Ok(rdata::MX::create(fb_builder, &args))
}

// TXT, from RFC1035 section 3.3.14
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                   TXT-DATA                    /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_txt(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let txt = &rdata.txt().context("missing rdata.txt")?;
    if let Some(entries) = txt.entries() {
        for i in 0..entries.len() {
            character_string::write(
                entries
                    .get(i)
                    .data()
                    .with_context(|| "missing txt entry data")?,
                buf,
                "rdata.txt.entries[]",
            )?;
        }
    }
    Ok(())
}

fn read_txt<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::TXT<'a>>> {
    let mut total_consumed: usize = 0;
    let mut entries = vec::Vec::new();
    while total_consumed < rdata_len {
        let (consumed, entry) =
            character_string::read(buf, rdata_offset + total_consumed, "rdata.txt[]")?;
        let args = rdata::TXTEntryArgs {
            data: Some(fb_builder.create_vector(entry.bytes())),
        };
        entries.push(rdata::TXTEntry::create(fb_builder, &args));
        total_consumed += consumed;
    }

    if total_consumed != rdata_len {
        bail!(
            "Expected TXT record data to be {} bytes, but was {} bytes",
            rdata_len,
            total_consumed
        );
    }

    let args = rdata::TXTArgs {
        entries: Some(fb_builder.create_vector(entries.as_slice())),
    };
    Ok(rdata::TXT::create(fb_builder, &args))
}

// RP, from RFC1183 section 2.2

fn write_rp(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    let rp = rdata.rp().context("missing rdata.rp")?;
    domain_name::write(
        rp.mbox_dname().context("missing rdata.rp.mbox_dname")?,
        buf,
        ptr_offsets,
        "rdata.rp.mbox_dname",
    )?;
    domain_name::write(
        rp.txt_dname().context("missing rdata.rp.txt_dname")?,
        buf,
        ptr_offsets,
        "rdata.rp.txt_dname",
    )?;
    Ok(())
}

fn read_rp<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::RP<'a>>> {
    // Each domain-name would need at least 1 byte for an initial size of 0
    if rdata_len < 2 {
        bail!(
            "RP record rdata requires at least {} bytes, got {}",
            2,
            rdata_len
        );
    }

    let mut offset = rdata_offset;

    let (mboxname_bytes_consumed, mboxname_str) =
        domain_name::read(buf, offset, "rdata.rp.mbox_dname")?;
    offset += mboxname_bytes_consumed;

    let (txtname_bytes_consumed, txtname_str) =
        domain_name::read(buf, offset, "rdata.rp.txt_dname")?;
    offset += txtname_bytes_consumed;

    if (offset - rdata_offset) != rdata_len {
        bail!(
            "Expected RP record data to be {} bytes, but was {} - {} bytes",
            rdata_len,
            offset,
            rdata_offset,
        );
    }

    let args = rdata::RPArgs {
        mbox_dname: Some(fb_builder.create_string(mboxname_str.as_str())),
        txt_dname: Some(fb_builder.create_string(txtname_str.as_str())),
    };
    Ok(rdata::RP::create(fb_builder, &args))
}

// AFSDB, from RFC1183 section 1

#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataAFSDBBits {
    #[packed_field(bits = "0:15")]
    subtype: u16,
    // hostname: variable length domain string
}

fn write_afsdb(
    rdata: &ResourceData,
    buf: &mut BytesMut,
    ptr_offsets: &mut domain_name::LabelOffsets,
) -> Result<()> {
    let afsdb = rdata.afsdb().context("missing rdata.afsdb")?;
    let packed_bits = RDataAFSDBBits {
        subtype: afsdb.subtype(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);

    domain_name::write(
        afsdb.hostname().context("missing rdata.afsdb.hostname")?,
        buf,
        ptr_offsets,
        "rdata.afsdb.hostname",
    )?;

    Ok(())
}

fn read_afsdb<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    mut rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::AFSDB<'a>>> {
    let bits_size = RDataAFSDBBits::packed_bytes();
    if rdata_len < bits_size + 1 {
        bail!(
            "AFSDB record rdata requires at least {} bytes, got {}",
            bits_size + 1,
            rdata_len
        );
    }
    let bits = RDataAFSDBBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack AFSDB record rdata bits")?;
    rdata_offset += bits_size;

    let (hostname_bytes_consumed, hostname_str) =
        domain_name::read(buf, rdata_offset, "rdata.afsdb.hostname")?;
    if bits_size + hostname_bytes_consumed != rdata_len {
        bail!(
            "Expected AFSDB record data to be {} bytes, but was {} + {} bytes",
            rdata_len,
            bits_size,
            hostname_bytes_consumed
        );
    }

    let args = rdata::AFSDBArgs {
        subtype: bits.subtype,
        hostname: Some(fb_builder.create_string(hostname_str.as_str())),
    };
    Ok(rdata::AFSDB::create(fb_builder, &args))
}

// AAAA, from RFC3596 (DIYed diagram: not included in spec)
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                                               |
// |                    ADDRESS                    |
// |                                               |
// |                                               |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+

#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataAAAABits {
    #[packed_field(bits = "0:15")]
    address1: u16,
    #[packed_field(bits = "16:31")]
    address2: u16,
    #[packed_field(bits = "32:47")]
    address3: u16,
    #[packed_field(bits = "48:63")]
    address4: u16,
    #[packed_field(bits = "64:79")]
    address5: u16,
    #[packed_field(bits = "80:95")]
    address6: u16,
    #[packed_field(bits = "96:111")]
    address7: u16,
    #[packed_field(bits = "112:127")]
    address8: u16,
}

pub fn write_aaaa_ip(ip: &Ipv6Addr, buf: &mut BytesMut) -> Result<()> {
    let segments = ip.segments();
    let packed_bits = RDataAAAABits {
        address1: segments[0],
        address2: segments[1],
        address3: segments[2],
        address4: segments[3],
        address5: segments[4],
        address6: segments[5],
        address7: segments[6],
        address8: segments[7],
    }
    .pack();
    buf.extend_from_slice(&packed_bits);
    Ok(())
}

fn write_aaaa(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let aaaa = rdata.aaaa().context("missing rdata.aaaa")?;
    let packed_bits = RDataAAAABits {
        address1: aaaa.address1(),
        address2: aaaa.address2(),
        address3: aaaa.address3(),
        address4: aaaa.address4(),
        address5: aaaa.address5(),
        address6: aaaa.address6(),
        address7: aaaa.address7(),
        address8: aaaa.address8(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);
    Ok(())
}

fn read_aaaa<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::AAAA<'a>>> {
    let bits_size = RDataAAAABits::packed_bytes();
    if rdata_len < bits_size {
        bail!(
            "AAAA record rdata requires at least {} bytes, got {}",
            bits_size,
            rdata_len
        );
    }
    let bits = RDataAAAABits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack AAAA record rdata bits")?;

    let args = rdata::AAAAArgs {
        address1: bits.address1,
        address2: bits.address2,
        address3: bits.address3,
        address4: bits.address4,
        address5: bits.address5,
        address6: bits.address6,
        address7: bits.address7,
        address8: bits.address8,
    };
    Ok(rdata::AAAA::create(fb_builder, &args))
}

// SRV, from RFC2782 (DIYed diagram: not included in spec)
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                   PRIORITY                    |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                    WEIGHT                     |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                     PORT                      |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                    TARGET                     /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataSRVBits {
    #[packed_field(bits = "0:15")]
    priority: u16,
    #[packed_field(bits = "16:31")]
    weight: u16,
    #[packed_field(bits = "32:47")]
    port: u16,
    // target: variable length domain string (no pointers)
}

fn write_srv(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let srv = rdata.srv().context("missing rdata.srv")?;
    let packed_bits = RDataSRVBits {
        priority: srv.priority(),
        weight: srv.weight(),
        port: srv.port(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);

    domain_name::write_nopointer(
        srv.target().context("missing rdata.srv.target")?,
        buf,
        "rdata.srv.target",
    )?;

    Ok(())
}

fn read_srv<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    mut rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::SRV<'a>>> {
    let bits_size = RDataSRVBits::packed_bytes();
    if rdata_len < bits_size + 1 {
        bail!(
            "SRV record rdata requires at least {} bytes, got {}",
            bits_size + 1,
            rdata_len
        );
    }
    let bits = RDataSRVBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack SRV record rdata bits")?;
    rdata_offset += bits_size;

    // RFC2782: "name compression is not to be used for this field."
    let (target_bytes_consumed, target_str) =
        domain_name::read_noptr(buf, rdata_offset, "rdata.srv.target")?;
    if bits_size + target_bytes_consumed != rdata_len {
        bail!(
            "Expected SRV record data to be {} bytes, but was {} + {} bytes",
            rdata_len,
            bits_size,
            target_bytes_consumed
        );
    }

    let args = rdata::SRVArgs {
        priority: bits.priority,
        weight: bits.weight,
        port: bits.priority,
        target: Some(fb_builder.create_string(target_str.as_str())),
    };
    Ok(rdata::SRV::create(fb_builder, &args))
}

// DNAME, from RFC6672 section 3.3.1
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// /                     CNAME                     /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
fn write_dname(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    domain_name::write_nopointer(
        rdata
            .dname()
            .context("missing rdata.dname")?
            .dname()
            .context("missing rdata.dname.dname")?,
        buf,
        "rdata.dname.dname",
    )?;
    Ok(())
}

fn read_dname<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::DNAME<'a>>> {
    let (name_bytes_consumed, name_str) =
        domain_name::read_noptr(buf, rdata_offset, "rdata.dname.dname")?;

    if name_bytes_consumed != rdata_len {
        bail!(
            "Expected DNAME record data to be {} bytes, but was {} bytes",
            rdata_len,
            name_bytes_consumed
        );
    }

    let args = rdata::DNAMEArgs {
        dname: Some(fb_builder.create_string(name_str.as_str())),
    };
    Ok(rdata::DNAME::create(fb_builder, &args))
}

// OPT rdata entry, from RFC6891 section 6.1.2 (MULTIPLE option entries like this)
//
//                                 1  1  1  1  1  1
//   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                  OPTION-CODE                  |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                 OPTION-LENGTH                 |
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
// |                                               |
// /                  OPTION-DATA                  /
// /                                               /
// +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataOPTEntryBits {
    #[packed_field(bits = "0:15")]
    code: u16,
    #[packed_field(bits = "16:31")]
    length: u16,
    // data: variable length content with meaning defined by the option code
}

/// Writes the list of entries to an OPT resource's rdata section, or does nothing if they're empty or unset
pub fn write_opt(opt: &OPT, buf: &mut BytesMut) -> Result<()> {
    if let Some(entries) = opt.option() {
        for i in 0..entries.len() {
            let entry = entries.get(i);
            let data = entry.data().context("rdata.opt.data")?;
            let packed_bits = RDataOPTEntryBits {
                code: entry.code(),
                length: u16::try_from(data.len()).with_context(|| "opt.data length doesn't fit")?,
            }
            .pack();
            buf.reserve(packed_bits.len() + data.len());
            buf.put_slice(&packed_bits);
            buf.put_slice(&data);
        }
    }
    Ok(())
}

/// Reads the list of entries from an OPT resource's rdata section, or an empty vec if there are none
pub fn read_opt<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
    opt_class_ttl_bits: message::ResourceClassTTLOPTBits,
) -> Result<flatbuffers::WIPOffset<OPT<'a>>> {
    let mut total_consumed: usize = 0;
    let mut entries = vec::Vec::new();

    // Extract OPT record entries. Each entry is preceded by a header that says how long that entry is.
    while total_consumed < rdata_len {
        let bits_size = RDataOPTEntryBits::packed_bytes();
        if rdata_len < total_consumed + bits_size {
            bail!(
                "OPT entry header requires at least {} bytes, got {}/{} consumed",
                bits_size,
                total_consumed,
                rdata_len
            );
        }
        let bits = RDataOPTEntryBits::unpack_from_slice(
            &buf[rdata_offset + total_consumed..rdata_offset + total_consumed + bits_size],
        )
        .context("couldn't unpack OPT record rdata bits")?;
        total_consumed += bits_size;

        if rdata_len < total_consumed + (bits.length as usize) {
            bail!(
                "OPT entry data requires at least {} bytes, got {}/{} consumed",
                bits.length,
                total_consumed,
                rdata_len
            );
        }

        // TODO(#21) support parsing the 'COOKIE' OPT option
        let args = OPTOptionArgs {
            code: bits.code,
            data: Some(fb_builder.create_vector(
                &buf[rdata_offset + total_consumed
                    ..rdata_offset + total_consumed + (bits.length as usize)],
            )),
        };
        entries.push(OPTOption::create(fb_builder, &args));
        total_consumed += bits.length as usize;
    }

    if total_consumed != rdata_len {
        bail!(
            "Expected OPT record data to be {} bytes, but was {} bytes",
            rdata_len,
            total_consumed
        );
    }

    let args = OPTArgs {
        option: Some(fb_builder.create_vector(entries.as_slice())),
        // Copy the OPT-specific class/ttl data into the OPT rdata
        udp_size: opt_class_ttl_bits.udp_size,
        response_code: opt_class_ttl_bits.response_code,
        version: opt_class_ttl_bits.version,
        dnssec_ok: opt_class_ttl_bits.dnssec_ok,
    };
    Ok(OPT::create(fb_builder, &args))
}

// DS, from RFC4034 section 5.1
//
//                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |           Key Tag             |  Algorithm    |  Digest Type  |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// /                                                               /
// /                            Digest                             /
// /                                                               /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataDSBits {
    #[packed_field(bits = "0:15")]
    key_tag: u16,
    #[packed_field(bits = "16:23")]
    algorithm: u8,
    #[packed_field(bits = "24:31")]
    digest_type: u8,
    // digest: variable length bytes
}

fn write_ds(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let ds = rdata.ds().context("missing rdata.ds")?;
    let packed_bits = RDataDSBits {
        key_tag: ds.key_tag(),
        algorithm: ds.algorithm(),
        digest_type: ds.digest_type(),
    }
    .pack();
    let digest = ds.digest().context("missing rdata.ds.digest")?;
    buf.reserve(packed_bits.len() + digest.len());
    buf.put_slice(&packed_bits);
    buf.put_slice(&digest);
    Ok(())
}

fn read_ds<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::DS<'a>>> {
    let bits_size = RDataDSBits::packed_bytes();
    if rdata_len < bits_size {
        bail!(
            "DS record rdata requires at least {} bytes, got {}",
            bits_size,
            rdata_len
        );
    }
    let bits = RDataDSBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack DS record rdata bits")?;

    let args = rdata::DSArgs {
        key_tag: bits.key_tag,
        algorithm: bits.algorithm,
        digest_type: bits.digest_type,
        // Remainder of the rdata is the digest
        digest: Some(
            fb_builder.create_vector(&buf[rdata_offset + bits_size..rdata_offset + rdata_len]),
        ),
    };
    Ok(rdata::DS::create(fb_builder, &args))
}

// RRSIG, from RFC4034 section 3.1
//
//                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |        Type Covered           |  Algorithm    |     Labels    |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |                         Original TTL                          |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |                      Signature Expiration                     |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |                      Signature Inception                      |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |            Key Tag            |                               /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+         Signer's Name         /
// /                                                               /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// /                                                               /
// /                            Signature                          /
// /                                                               /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataRRSIGBits {
    #[packed_field(bits = "0:15")]
    type_covered: u16,
    #[packed_field(bits = "16:23")]
    algorithm: u8,
    #[packed_field(bits = "24:31")]
    labels: u8,
    #[packed_field(bits = "32:63")]
    original_ttl: u32,
    #[packed_field(bits = "64:95")]
    signature_expiration: u32,
    #[packed_field(bits = "96:127")]
    signature_inception: u32,
    #[packed_field(bits = "128:143")]
    key_tag: u16,
    // signer's name: domain string, no compression
    // signature: variable bytes (rest of rdata)
}

fn write_rrsig(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let rrsig = rdata.rrsig().context("missing rdata.rrsig")?;
    let packed_bits = RDataRRSIGBits {
        type_covered: rrsig.type_covered(),
        algorithm: rrsig.algorithm(),
        labels: rrsig.labels(),
        original_ttl: rrsig.original_ttl(),
        signature_expiration: rrsig.signature_expiration(),
        signature_inception: rrsig.signature_inception(),
        key_tag: rrsig.key_tag(),
    }
    .pack();
    buf.extend_from_slice(&packed_bits);
    domain_name::write_nopointer(
        rrsig
            .signers_name()
            .context("missing rdata.rrsig.signers_name")?,
        buf,
        "rdata.rrsig.signers_name",
    )?;
    buf.extend_from_slice(rrsig.signature().context("missing rdata.rrsig.signature")?);
    Ok(())
}

fn read_rrsig<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::RRSIG<'a>>> {
    let bits_size = RDataRRSIGBits::packed_bytes();
    if rdata_len < bits_size {
        bail!(
            "RRSIG record rdata requires at least {} bytes, got {}",
            bits_size,
            rdata_len
        );
    }
    let bits = RDataRRSIGBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack RRSIG record rdata bits")?;
    let mut total_consumed = bits_size;

    let (signers_name_bytes_consumed, signers_name_str) = domain_name::read_noptr(
        buf,
        rdata_offset + total_consumed,
        "rdata.rrsig.signers_name",
    )?;
    total_consumed += signers_name_bytes_consumed;

    let args = rdata::RRSIGArgs {
        type_covered: bits.type_covered,
        algorithm: bits.algorithm,
        labels: bits.labels,
        original_ttl: bits.original_ttl,
        signature_expiration: bits.signature_expiration,
        signature_inception: bits.signature_inception,
        key_tag: bits.key_tag,
        signers_name: Some(fb_builder.create_string(signers_name_str.as_str())),
        // Remainder of the rdata is the signature
        signature: Some(
            fb_builder.create_vector(&buf[rdata_offset + total_consumed..rdata_offset + rdata_len]),
        ),
    };
    Ok(rdata::RRSIG::create(fb_builder, &args))
}

// NSEC, from RFC4034 section 4.1
//
//                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// /                      Next Domain Name                         /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// /                       Type Bit Maps                           /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
fn write_nsec(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let nsec = rdata.nsec().context("missing rdata.nsec")?;
    domain_name::write_nopointer(
        nsec.next_domain_name()
            .context("missing rdata.nsec.next_domain_name")?,
        buf,
        "rdata.nsec.next_domain_name",
    )?;
    buf.extend_from_slice(
        nsec.type_bit_maps()
            .context("missing rdata.nsec.type_bit_maps")?,
    );
    Ok(())
}

fn read_nsec<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::NSEC<'a>>> {
    let (next_domain_bytes_consumed, next_domain_str) =
        domain_name::read_noptr(buf, rdata_offset, "rdata.nsec.next_domain_name")?;
    let args = rdata::NSECArgs {
        next_domain_name: Some(fb_builder.create_string(next_domain_str.as_str())),
        // Remainder of the rdata is the bitmaps
        type_bit_maps: Some(fb_builder.create_vector(
            &buf[rdata_offset + next_domain_bytes_consumed..rdata_offset + rdata_len],
        )),
    };
    Ok(rdata::NSEC::create(fb_builder, &args))
}

// DNSKEY, from RFC4034 section 2.1
//
//                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// |              Flags            |    Protocol   |   Algorithm   |
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
// /                                                               /
// /                            Public Key                         /
// /                                                               /
// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
#[derive(PackedStruct)]
#[packed_struct(endian = "msb", bit_numbering = "msb0")]
pub struct RDataDNSKEYBits {
    #[packed_field(bits = "0:15")]
    flags: u16,
    #[packed_field(bits = "16:23")]
    protocol: u8,
    #[packed_field(bits = "24:31")]
    algorithm: u8,
    // public key: variable bytes (rest of rdata)
}

fn write_dnskey(rdata: &ResourceData, buf: &mut BytesMut) -> Result<()> {
    let dnskey = rdata.dnskey().context("missing rdata.dnskey")?;
    let packed_bits = RDataDNSKEYBits {
        flags: dnskey.flags(),
        protocol: dnskey.protocol(),
        algorithm: dnskey.algorithm(),
    }
    .pack();
    let public_key = dnskey
        .public_key()
        .context("missing rdata.dnskey.public_key")?;
    buf.reserve(packed_bits.len() + public_key.len());
    buf.put_slice(&packed_bits);
    buf.put_slice(&public_key);
    Ok(())
}

fn read_dnskey<'a>(
    fb_builder: &mut flatbuffers::FlatBufferBuilder<'a>,
    buf: &[u8],
    rdata_offset: usize,
    rdata_len: usize,
) -> Result<flatbuffers::WIPOffset<rdata::DNSKEY<'a>>> {
    let bits_size = RDataDNSKEYBits::packed_bytes();
    if rdata_len < bits_size {
        bail!(
            "DNSKEY record rdata requires at least {} bytes, got {}",
            bits_size,
            rdata_len
        );
    }
    let bits = RDataDNSKEYBits::unpack_from_slice(&buf[rdata_offset..rdata_offset + bits_size])
        .context("couldn't unpack DNSKEY record rdata bits")?;

    let args = rdata::DNSKEYArgs {
        flags: bits.flags,
        protocol: bits.protocol,
        algorithm: bits.algorithm,
        // Remainder of the rdata is the public key
        public_key: Some(
            fb_builder.create_vector(&buf[rdata_offset + bits_size..rdata_offset + rdata_len]),
        ),
    };
    Ok(rdata::DNSKEY::create(fb_builder, &args))
}