// MosquittoClient against a real broker (docs/DESIGN.md, D30 level 2): a
// local mosquitto on a free port. Connect, messages, retained messages,
// reconnect, last will, login. Skips when mosquitto isn't installed.
//
// The binary is also its own helper process for the last-will test:
// `tst_mqtt --will-helper <port> <topic>` connects with a will and waits to be
// killed.

#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <QCoreApplication>
#include <QFile>
#include <QProcess>
#include <QStandardPaths>
#include <QTemporaryDir>
#include <QTest>
#include <memory>

#include "mqtt/MosquittoClient.h"

namespace {

constexpr int kTimeoutMs = 10000;

QString findTool(const QString& name) {
  QString path = QStandardPaths::findExecutable(name);
  if (path.isEmpty())
    path = QStandardPaths::findExecutable(name, {QStringLiteral("/usr/sbin")});
  return path;
}

// A TCP port nobody listens on right now.
int freePort() {
  const int fd = socket(AF_INET, SOCK_STREAM, 0);
  sockaddr_in address{};
  address.sin_family = AF_INET;
  address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
  socklen_t length = sizeof(address);
  bind(fd, reinterpret_cast<sockaddr*>(&address), length);
  getsockname(fd, reinterpret_cast<sockaddr*>(&address), &length);
  close(fd);
  return ntohs(address.sin_port);
}

bool portOpen(int port) {
  const int fd = socket(AF_INET, SOCK_STREAM, 0);
  sockaddr_in address{};
  address.sin_family = AF_INET;
  address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
  address.sin_port = htons(port);
  const bool open = ::connect(fd, reinterpret_cast<sockaddr*>(&address),
                              sizeof(address)) == 0;
  close(fd);
  return open;
}

// A mosquitto process with its own configuration.
class Broker {
 public:
  explicit Broker(const QString& dir) : m_dir(dir), m_port(freePort()) {}
  ~Broker() { stop(); }

  int port() const { return m_port; }

  // `extra`: further configuration lines; without them anonymous clients may
  // connect.
  bool start(const QString& extra = QStringLiteral("allow_anonymous true\n")) {
    const QString config = m_dir + QStringLiteral("/mosquitto.conf");
    QFile file(config);
    if (!file.open(QIODevice::WriteOnly)) return false;
    file.write(QStringLiteral("listener %1 127.0.0.1\npersistence false\n%2")
                   .arg(m_port)
                   .arg(extra)
                   .toUtf8());
    file.close();
    m_process = std::make_unique<QProcess>();
    m_process->setProcessChannelMode(QProcess::ForwardedErrorChannel);
    m_process->start(findTool(QStringLiteral("mosquitto")),
                     {QStringLiteral("-c"), config});
    if (!m_process->waitForStarted()) return false;
    QDeadlineTimer deadline(5000);
    while (!portOpen(m_port)) {
      if (deadline.hasExpired() || m_process->state() != QProcess::Running)
        return false;
      QThread::msleep(20);
    }
    return true;
  }

  void stop() {
    if (!m_process) return;
    m_process->terminate();
    if (!m_process->waitForFinished(5000)) m_process->kill();
    m_process->waitForFinished();
    m_process.reset();
  }

 private:
  QString m_dir;
  int m_port;
  std::unique_ptr<QProcess> m_process;
};

MosquittoClient::Settings settings(int port, const QString& clientId) {
  MosquittoClient::Settings s;
  s.broker.host = QStringLiteral("127.0.0.1");
  s.broker.port = port;
  s.clientId = clientId;
  s.reconnectMinMs = 100;
  s.reconnectMaxMs = 500;
  return s;
}

// Collects a client's signals in the main thread (queued connections).
class Recorder : public QObject {
 public:
  explicit Recorder(MqttClient* client) {
    connect(client, &MqttClient::connectionChanged, this,
            [this](MqttClient::State state, const QString& error) {
              states.append(state);
              if (!error.isEmpty()) errors.append(error);
            });
    connect(client, &MqttClient::messageReceived, this,
            [this](const QString& topic, const QByteArray& payload) {
              messages.append({topic, payload});
            });
  }

  QList<MqttClient::State> states;
  QStringList errors;
  QList<QPair<QString, QByteArray>> messages;
};

// Connects and waits for it; then subscribes to `topic` if given.
bool connectClient(MosquittoClient& client, Recorder& recorder,
                   const QString& topic = {}) {
  client.connectToBroker();
  if (!QTest::qWaitFor(
          [&] {
            return recorder.states.contains(MqttClient::State::Connected);
          },
          kTimeoutMs))
    return false;
  if (topic.isEmpty()) return true;
  if (!client.subscribe(topic)) return false;
  // No SUBACK signal: give the broker a moment to register it.
  QTest::qWait(200);
  return true;
}

int runWillHelper(int argc, char** argv) {
  QCoreApplication app(argc, argv);
  MosquittoClient::Settings s =
      settings(QByteArray(argv[2]).toInt(), QStringLiteral("will-helper"));
  s.will = {QString::fromUtf8(argv[3]), "offline", true};
  MosquittoClient client(s);
  QObject::connect(&client, &MqttClient::connectionChanged, &app,
                   [](MqttClient::State state) {
                     if (state == MqttClient::State::Connected) {
                       fputs("connected\n", stdout);
                       fflush(stdout);
                     }
                   });
  client.connectToBroker();
  return app.exec();
}

}  // namespace

class TestMqtt : public QObject {
  Q_OBJECT

 private slots:
  void initTestCase() {
    if (findTool(QStringLiteral("mosquitto")).isEmpty())
      QSKIP("mosquitto (the broker) is not installed");
    QVERIFY(m_dir.isValid());
  }

  void connectsAndExchangesMessages() {
    Broker broker(m_dir.path());
    QVERIFY(broker.start());
    MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
    MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
    Recorder ra(&a), rb(&b);
    QVERIFY(connectClient(a, ra));
    QCOMPARE(ra.states.first(), MqttClient::State::Connecting);
    QCOMPARE(a.state(), MqttClient::State::Connected);
    QVERIFY(connectClient(b, rb, QStringLiteral("test/#")));

    QVERIFY(a.publish(QStringLiteral("test/x"), "hello", false));
    QVERIFY(a.waitForPublished(kTimeoutMs));
    QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
    QCOMPARE(rb.messages.first().first, QStringLiteral("test/x"));
    QCOMPARE(rb.messages.first().second, QByteArray("hello"));
    QVERIFY(ra.errors.isEmpty());
  }

  void retainedMessageReachesLaterSubscriber() {
    Broker broker(m_dir.path());
    QVERIFY(broker.start());
    MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
    Recorder ra(&a);
    QVERIFY(connectClient(a, ra));
    QVERIFY(a.publish(QStringLiteral("test/retained"), "kept", true));
    QVERIFY(a.waitForPublished(kTimeoutMs));

    MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
    Recorder rb(&b);
    QVERIFY(connectClient(b, rb, QStringLiteral("test/retained")));
    QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
    QCOMPARE(rb.messages.first().second, QByteArray("kept"));
  }

  void publishFailsWhileDisconnected() {
    MosquittoClient a(settings(freePort(), QStringLiteral("a")));
    QVERIFY(!a.publish(QStringLiteral("test/x"), "lost", false));
    QVERIFY(!a.subscribe(QStringLiteral("test/x")));
    QVERIFY(a.waitForPublished(0));
  }

  // The broker isn't running yet: the error is reported, and the client
  // connects once it runs.
  void firstConnectRetriesUntilBrokerRuns() {
    Broker broker(m_dir.path());
    MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
    Recorder ra(&a);
    a.connectToBroker();
    QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
    QCOMPARE(a.state(), MqttClient::State::Disconnected);
    QVERIFY2(ra.errors.first().contains(QStringLiteral("refused")),
             qPrintable(ra.errors.first()));
    QVERIFY(broker.start());
    QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
                              kTimeoutMs);
  }

  void reconnectsAfterBrokerRestart() {
    Broker broker(m_dir.path());
    QVERIFY(broker.start());
    MosquittoClient a(settings(broker.port(), QStringLiteral("a")));
    Recorder ra(&a);
    QVERIFY(connectClient(a, ra));
    broker.stop();
    QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Disconnected,
                              kTimeoutMs);
    QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
    ra.states.clear();
    QVERIFY(broker.start());
    QTRY_VERIFY_WITH_TIMEOUT(ra.states.contains(MqttClient::State::Connected),
                             kTimeoutMs);
  }

  void unresolvableHostIsReported() {
    MosquittoClient::Settings s = settings(1883, QStringLiteral("a"));
    s.broker.host = QStringLiteral("no-such-host.invalid");
    MosquittoClient a(s);
    Recorder ra(&a);
    a.connectToBroker();
    QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
    QVERIFY2(ra.errors.first().startsWith(QStringLiteral("Cannot resolve")),
             qPrintable(ra.errors.first()));
  }

  // CON-3: a killed process (no DISCONNECT) → the broker publishes the will.
  void lastWillWhenKilled() {
    Broker broker(m_dir.path());
    QVERIFY(broker.start());
    MosquittoClient watcher(settings(broker.port(), QStringLiteral("w")));
    Recorder rw(&watcher);
    QVERIFY(connectClient(watcher, rw, QStringLiteral("test/availability")));

    QProcess helper;
    helper.start(
        QCoreApplication::applicationFilePath(),
        {QStringLiteral("--will-helper"), QString::number(broker.port()),
         QStringLiteral("test/availability")});
    QVERIFY(helper.waitForStarted());
    // QTRY_* evaluates its expression again at the end: collect the output.
    QByteArray output;
    QTRY_VERIFY_WITH_TIMEOUT(
        (output += helper.readAllStandardOutput()).contains("connected"),
        kTimeoutMs);
    QVERIFY(rw.messages.isEmpty());
    helper.kill();
    helper.waitForFinished();
    QTRY_COMPARE_WITH_TIMEOUT(rw.messages.size(), 1, kTimeoutMs);
    QCOMPARE(rw.messages.first().second, QByteArray("offline"));
  }

  // A normal exit publishes the will itself (retained), then disconnects.
  void disconnectPublishesWill() {
    Broker broker(m_dir.path());
    QVERIFY(broker.start());
    MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a"));
    s.will = {QStringLiteral("test/availability"), "offline", true};
    MosquittoClient a(s);
    Recorder ra(&a);
    QVERIFY(connectClient(a, ra));
    QVERIFY(a.disconnectFromBroker(kTimeoutMs));
    QCOMPARE(a.state(), MqttClient::State::Disconnected);
    QTRY_COMPARE_WITH_TIMEOUT(ra.states.last(), MqttClient::State::Disconnected,
                              kTimeoutMs);

    MosquittoClient b(settings(broker.port(), QStringLiteral("b")));
    Recorder rb(&b);
    QVERIFY(connectClient(b, rb, QStringLiteral("test/availability")));
    QTRY_COMPARE_WITH_TIMEOUT(rb.messages.size(), 1, kTimeoutMs);
    QCOMPARE(rb.messages.first().second, QByteArray("offline"));
  }

  void login_data() {
    QTest::addColumn<QString>("password");
    QTest::addColumn<bool>("accepted");
    QTest::newRow("right") << QStringLiteral("secret") << true;
    QTest::newRow("wrong") << QStringLiteral("wrong") << false;
  }

  void login() {
    QFETCH(QString, password);
    QFETCH(bool, accepted);
    const QString passwd = findTool(QStringLiteral("mosquitto_passwd"));
    if (passwd.isEmpty()) QSKIP("mosquitto_passwd is not installed");
    const QString file = m_dir.path() + QStringLiteral("/passwd");
    QFile::remove(file);
    QCOMPARE(QProcess::execute(
                 passwd, {QStringLiteral("-c"), QStringLiteral("-b"), file,
                          QStringLiteral("frame"), QStringLiteral("secret")}),
             0);
    Broker broker(m_dir.path());
    QVERIFY(broker.start(
        QStringLiteral("allow_anonymous false\npassword_file %1\n").arg(file)));

    MosquittoClient::Settings s = settings(broker.port(), QStringLiteral("a"));
    s.broker.username = QStringLiteral("frame");
    s.broker.password = password;
    MosquittoClient a(s);
    Recorder ra(&a);
    a.connectToBroker();
    if (accepted) {
      QTRY_COMPARE_WITH_TIMEOUT(a.state(), MqttClient::State::Connected,
                                kTimeoutMs);
    } else {
      QTRY_VERIFY_WITH_TIMEOUT(!ra.errors.isEmpty(), kTimeoutMs);
      QVERIFY2(ra.errors.first().contains(QStringLiteral("not authori")),
               qPrintable(ra.errors.first()));
      QVERIFY(!ra.states.contains(MqttClient::State::Connected));
    }
  }

 private:
  QTemporaryDir m_dir;
};

int main(int argc, char** argv) {
  if (argc == 4 && qstrcmp(argv[1], "--will-helper") == 0)
    return runWillHelper(argc, argv);
  QCoreApplication app(argc, argv);
  TestMqtt test;
  return QTest::qExec(&test, argc, argv);
}

#include "tst_mqtt.moc"