alvi31182v

GROUP BY GROUP_CONCAT для объединения сообщений в одну строку, упорядоченную по времени т.е если у тебя есть поле timestamp или msg_time если ты не хочешь использовать chat_id, то можно попробовать сделать следующий запрос

GROUP BY chat_id - таким образом, строки с одинаковым chat_id будут объединены в одну группу

"Функция LEAST выбирает минимальное из значений. Значениями могут выступать поля, а также строки и числа"
"Функция GREATEST выбирает максимальное из значений / помогает упорядочивать пары пользователей"

Комбинация LEAST и GREATEST гарантирует уникальность групировки чатов игнорируя порядок пар пользоваталей чатов.
Вобщем есть набросок, дальше можно уже самому накидать и поиграться.

$sql = "
    SELECT 
        MIN(from_user_id) as user1,
        MAX(from_user_id) as user2,
        chat_id,
        GROUP_CONCAT(msg ORDER BY msg_time) as messages
    FROM
        class_chat
    WHERE
        (from_user_id = :user1 AND to_user_id = :user2)
        OR
        (to_user_id = :user1 AND from_user_id = :user2)
        OR
        (from_user_id = :user3 AND to_user_id = :user4)
        OR
        (to_user_id = :user3 AND from_user_id = :user4)
    GROUP BY
        chat_id, LEAST(from_user_id, to_user_id), GREATEST(from_user_id, to_user_id)
";

$stmt = $pdo->prepare($sql);

$stmt->execute([
    ':user1' => 111, // тут айдишники пользаков
    ':user2' => 222,
    ':user3' => 333,
    ':user4' => 444,
]);

while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
    // Обработка результата
    echo "Chat ID: " . $row['chat_id'] . "\n";
    echo "User 1: " . $row['user1'] . "\n";
    echo "User 2: " . $row['user2'] . "\n";
    echo "Messages: " . $row['messages'] . "\n";
    echo "-----------------\n";
}

Нужно отметить что сам fopen это ведь встроенная функция в php и мокать её напрямую достаточно сложно,
Для этого можно имплементировать типа FileReaderInterface и использовать DI.

interface FileOpenerInterface {
    public function openFile($path, $mode);
}

final class FileOpener implements  FileOpenerInterface{
public function openFile($path, $mode){
return fopen($path, $mode);
}
}

class FileProcessor{

 public function __construct(private FileOpenerInterface $fileOpener) { }

public function readFileContent($filePath) {
        $fileHandle = $this->fileOpener->openFile($filePath, 'r');
        
        if ($fileHandle) {
            $content = fread($fileHandle, filesize($filePath));
            fclose($fileHandle);
            return $content;
        } else {
            return false;
        }
    }
}

class FileProcessorTest extends PHPUnit\Framework\TestCase {

    public function testReadFileContent() {
        // Создаем мок для FileOpenerInterface
        $fileOpenerMock = $this->createMock(FileOpenerInterface::class);


        $fileOpenerMock->expects($this->once())
            ->method('openFile')
            ->willReturn(fopen('php://memory', 'r'));


        $fileProcessor = new FileProcessor($fileOpenerMock);
        $content = $fileProcessor->readFileContent('fakefile://path/to/your/file.txt');

        // Проверяем ожидаемый результат
        $this->assertEquals("", $content);
    }
}

Сам этот тест я не зпускал но для понимания думаю пойдет, там дальше уже можешь сам что нибудь накидать, прикинуть что тебе нужно и т.д

Не знаю как тут в amp php
но есть такой интерфейс как https://amphp.org/parallel#tasks
В reactPHP есть promise в которые ты передаешь через
Deferred в метод resolve(данные которые тебе надо обработать) : PromiseInterface.
Попробуй в RunContentJobTask имплементировать Task и запустить в методе run выполнения метода save.
не знаю может у тебя уже так и сделано.

https://github.com/arnaud-lb/php-rdkafka - на данный момент это пока самая рабочая либа. И без глубокого понимания как работает кафка тебе не удасться сделать клиент учитывая все абсолютно все настройки для кафки.
Через композер тебе только пакет (kwn/php-rdkafka-stubs - "Rdkafka extension stubs for your IDE"). Остальное уже librdkafka (pecl) и на оф manual для запуска циклов опроса для чтения данных из кафки есть примеры и они хочешь не хочешь завязаны уже в pecl
https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc...

Если тебя волнует проблема безопасности этих либ и внутрянки с уязвимостями то просто сделай
composer audit
никаких проблем с бозопасностью

gzcompress сжимает данные с использованием алгоритма сжатия Zlib

Приблизительно можно посчитать сколько символов будет на выходе

$data = "Твои данные для сжатия"; 

$compressed_data = gzcompress($data);
$base64_encoded_data = base64_encode($compressed_data);

echo "Исходный размер: " . strlen($data) . " байт\n";
echo "Размер после сжатия: " . strlen($compressed_data) . " байт\n";
echo "Размер после base64-кодирования: " . strlen($base64_encoded_data) . " символов\n";

пых не умеет асинхронно перезагружать страницу , тебе надо заюзать js либу из самого простого это jQuery