haskell-pgmq (empty) → 0.1.0.0
raw patch · 16 files changed
+2582/−0 lines, 16 filesdep +aesondep +basedep +bytestring
Dependencies added: aeson, base, bytestring, containers, haskell-pgmq, hspec, postgresql-simple, random-strings, safe, stm, tasty, tasty-hspec, text, time, units
Files
- CHANGELOG.md +28/−0
- LICENSE +661/−0
- README.md +39/−0
- bin/simple-exception-catch/Main.hs +31/−0
- bin/simple-test/Main.hs +90/−0
- haskell-pgmq.cabal +205/−0
- sql/pgmq.sql +778/−0
- src/Database/PGMQ.hs +20/−0
- src/Database/PGMQ/Helpers.hs +34/−0
- src/Database/PGMQ/Schema.hs +9/−0
- src/Database/PGMQ/Simple.hs +311/−0
- src/Database/PGMQ/Types.hs +120/−0
- tests/Test/Integration/PGMQ/Helpers.hs +70/−0
- tests/Test/Integration/PGMQ/Simple.hs +138/−0
- tests/Test/Integration/Utils.hs +28/−0
- tests/integration-tests.hs +20/−0
+ CHANGELOG.md view
@@ -0,0 +1,28 @@+# Changelog++All notable changes to this project will be documented in this file.++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).++## [Unreleased]++## [0.1.0.0] - 2024-XX-XX++### Added+- Initial Haskell interface for Tembo's PGMQ PostgreSQL extension+- Core modules: `Database.PGMQ`, `Database.PGMQ.Simple`, `Database.PGMQ.Types`+- Helper modules: `Database.PGMQ.Helpers`, `Database.PGMQ.Schema`+- Simple test executable (`simple-test`) for basic functionality testing+- Simple exception catching executable (`simple-exception-catch`)+- Integration test suite with PostgreSQL+- Support for PostgreSQL message queue operations+- Documentation and examples in README++### Dependencies+- postgresql-simple for database connectivity+- aeson for JSON handling+- Various supporting libraries (bytestring, text, time, units, safe)++[Unreleased]: https://gitlab.iscpif.fr/gargantext/haskell-pgmq/compare/v0.1.0.0...HEAD+[0.1.0.0]: https://gitlab.iscpif.fr/gargantext/haskell-pgmq/releases/tag/v0.1.0.0
+ LICENSE view
@@ -0,0 +1,661 @@+ GNU AFFERO GENERAL PUBLIC LICENSE+ Version 3, 19 November 2007++ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.++ Preamble++ The GNU Affero General Public License is a free, copyleft license for+software and other kinds of works, specifically designed to ensure+cooperation with the community in the case of network server software.++ The licenses for most software and other practical works are designed+to take away your freedom to share and change the works. By contrast,+our General Public Licenses are intended to guarantee your freedom to+share and change all versions of a program--to make sure it remains free+software for all its users.++ When we speak of free software, we are referring to freedom, not+price. Our General Public Licenses are designed to make sure that you+have the freedom to distribute copies of free software (and charge for+them if you wish), that you receive source code or can get it if you+want it, that you can change the software or use pieces of it in new+free programs, and that you know you can do these things.++ Developers that use our General Public Licenses protect your rights+with two steps: (1) assert copyright on the software, and (2) offer+you this License which gives you legal permission to copy, distribute+and/or modify the software.++ A secondary benefit of defending all users' freedom is that+improvements made in alternate versions of the program, if they+receive widespread use, become available for other developers to+incorporate. Many developers of free software are heartened and+encouraged by the resulting cooperation. However, in the case of+software used on network servers, this result may fail to come about.+The GNU General Public License permits making a modified version and+letting the public access it on a server without ever releasing its+source code to the public.++ The GNU Affero General Public License is designed specifically to+ensure that, in such cases, the modified source code becomes available+to the community. It requires the operator of a network server to+provide the source code of the modified version running there to the+users of that server. Therefore, public use of a modified version, on+a publicly accessible server, gives the public access to the source+code of the modified version.++ An older license, called the Affero General Public License and+published by Affero, was designed to accomplish similar goals. This is+a different license, not a version of the Affero GPL, but Affero has+released a new version of the Affero GPL which permits relicensing under+this license.++ The precise terms and conditions for copying, distribution and+modification follow.++ TERMS AND CONDITIONS++ 0. Definitions.++ "This License" refers to version 3 of the GNU Affero General Public License.++ "Copyright" also means copyright-like laws that apply to other kinds of+works, such as semiconductor masks.++ "The Program" refers to any copyrightable work licensed under this+License. Each licensee is addressed as "you". "Licensees" and+"recipients" may be individuals or organizations.++ To "modify" a work means to copy from or adapt all or part of the work+in a fashion requiring copyright permission, other than the making of an+exact copy. The resulting work is called a "modified version" of the+earlier work or a work "based on" the earlier work.++ A "covered work" means either the unmodified Program or a work based+on the Program.++ To "propagate" a work means to do anything with it that, without+permission, would make you directly or secondarily liable for+infringement under applicable copyright law, except executing it on a+computer or modifying a private copy. Propagation includes copying,+distribution (with or without modification), making available to the+public, and in some countries other activities as well.++ To "convey" a work means any kind of propagation that enables other+parties to make or receive copies. Mere interaction with a user through+a computer network, with no transfer of a copy, is not conveying.++ An interactive user interface displays "Appropriate Legal Notices"+to the extent that it includes a convenient and prominently visible+feature that (1) displays an appropriate copyright notice, and (2)+tells the user that there is no warranty for the work (except to the+extent that warranties are provided), that licensees may convey the+work under this License, and how to view a copy of this License. If+the interface presents a list of user commands or options, such as a+menu, a prominent item in the list meets this criterion.++ 1. Source Code.++ The "source code" for a work means the preferred form of the work+for making modifications to it. "Object code" means any non-source+form of a work.++ A "Standard Interface" means an interface that either is an official+standard defined by a recognized standards body, or, in the case of+interfaces specified for a particular programming language, one that+is widely used among developers working in that language.++ The "System Libraries" of an executable work include anything, other+than the work as a whole, that (a) is included in the normal form of+packaging a Major Component, but which is not part of that Major+Component, and (b) serves only to enable use of the work with that+Major Component, or to implement a Standard Interface for which an+implementation is available to the public in source code form. A+"Major Component", in this context, means a major essential component+(kernel, window system, and so on) of the specific operating system+(if any) on which the executable work runs, or a compiler used to+produce the work, or an object code interpreter used to run it.++ The "Corresponding Source" for a work in object code form means all+the source code needed to generate, install, and (for an executable+work) run the object code and to modify the work, including scripts to+control those activities. However, it does not include the work's+System Libraries, or general-purpose tools or generally available free+programs which are used unmodified in performing those activities but+which are not part of the work. For example, Corresponding Source+includes interface definition files associated with source files for+the work, and the source code for shared libraries and dynamically+linked subprograms that the work is specifically designed to require,+such as by intimate data communication or control flow between those+subprograms and other parts of the work.++ The Corresponding Source need not include anything that users+can regenerate automatically from other parts of the Corresponding+Source.++ The Corresponding Source for a work in source code form is that+same work.++ 2. Basic Permissions.++ All rights granted under this License are granted for the term of+copyright on the Program, and are irrevocable provided the stated+conditions are met. This License explicitly affirms your unlimited+permission to run the unmodified Program. The output from running a+covered work is covered by this License only if the output, given its+content, constitutes a covered work. This License acknowledges your+rights of fair use or other equivalent, as provided by copyright law.++ You may make, run and propagate covered works that you do not+convey, without conditions so long as your license otherwise remains+in force. You may convey covered works to others for the sole purpose+of having them make modifications exclusively for you, or provide you+with facilities for running those works, provided that you comply with+the terms of this License in conveying all material for which you do+not control copyright. Those thus making or running the covered works+for you must do so exclusively on your behalf, under your direction+and control, on terms that prohibit them from making any copies of+your copyrighted material outside their relationship with you.++ Conveying under any other circumstances is permitted solely under+the conditions stated below. Sublicensing is not allowed; section 10+makes it unnecessary.++ 3. Protecting Users' Legal Rights From Anti-Circumvention Law.++ No covered work shall be deemed part of an effective technological+measure under any applicable law fulfilling obligations under article+11 of the WIPO copyright treaty adopted on 20 December 1996, or+similar laws prohibiting or restricting circumvention of such+measures.++ When you convey a covered work, you waive any legal power to forbid+circumvention of technological measures to the extent such circumvention+is effected by exercising rights under this License with respect to+the covered work, and you disclaim any intention to limit operation or+modification of the work as a means of enforcing, against the work's+users, your or third parties' legal rights to forbid circumvention of+technological measures.++ 4. Conveying Verbatim Copies.++ You may convey verbatim copies of the Program's source code as you+receive it, in any medium, provided that you conspicuously and+appropriately publish on each copy an appropriate copyright notice;+keep intact all notices stating that this License and any+non-permissive terms added in accord with section 7 apply to the code;+keep intact all notices of the absence of any warranty; and give all+recipients a copy of this License along with the Program.++ You may charge any price or no price for each copy that you convey,+and you may offer support or warranty protection for a fee.++ 5. Conveying Modified Source Versions.++ You may convey a work based on the Program, or the modifications to+produce it from the Program, in the form of source code under the+terms of section 4, provided that you also meet all of these conditions:++ a) The work must carry prominent notices stating that you modified+ it, and giving a relevant date.++ b) The work must carry prominent notices stating that it is+ released under this License and any conditions added under section+ 7. This requirement modifies the requirement in section 4 to+ "keep intact all notices".++ c) You must license the entire work, as a whole, under this+ License to anyone who comes into possession of a copy. This+ License will therefore apply, along with any applicable section 7+ additional terms, to the whole of the work, and all its parts,+ regardless of how they are packaged. This License gives no+ permission to license the work in any other way, but it does not+ invalidate such permission if you have separately received it.++ d) If the work has interactive user interfaces, each must display+ Appropriate Legal Notices; however, if the Program has interactive+ interfaces that do not display Appropriate Legal Notices, your+ work need not make them do so.++ A compilation of a covered work with other separate and independent+works, which are not by their nature extensions of the covered work,+and which are not combined with it such as to form a larger program,+in or on a volume of a storage or distribution medium, is called an+"aggregate" if the compilation and its resulting copyright are not+used to limit the access or legal rights of the compilation's users+beyond what the individual works permit. Inclusion of a covered work+in an aggregate does not cause this License to apply to the other+parts of the aggregate.++ 6. Conveying Non-Source Forms.++ You may convey a covered work in object code form under the terms+of sections 4 and 5, provided that you also convey the+machine-readable Corresponding Source under the terms of this License,+in one of these ways:++ a) Convey the object code in, or embodied in, a physical product+ (including a physical distribution medium), accompanied by the+ Corresponding Source fixed on a durable physical medium+ customarily used for software interchange.++ b) Convey the object code in, or embodied in, a physical product+ (including a physical distribution medium), accompanied by a+ written offer, valid for at least three years and valid for as+ long as you offer spare parts or customer support for that product+ model, to give anyone who possesses the object code either (1) a+ copy of the Corresponding Source for all the software in the+ product that is covered by this License, on a durable physical+ medium customarily used for software interchange, for a price no+ more than your reasonable cost of physically performing this+ conveying of source, or (2) access to copy the+ Corresponding Source from a network server at no charge.++ c) Convey individual copies of the object code with a copy of the+ written offer to provide the Corresponding Source. This+ alternative is allowed only occasionally and noncommercially, and+ only if you received the object code with such an offer, in accord+ with subsection 6b.++ d) Convey the object code by offering access from a designated+ place (gratis or for a charge), and offer equivalent access to the+ Corresponding Source in the same way through the same place at no+ further charge. You need not require recipients to copy the+ Corresponding Source along with the object code. If the place to+ copy the object code is a network server, the Corresponding Source+ may be on a different server (operated by you or a third party)+ that supports equivalent copying facilities, provided you maintain+ clear directions next to the object code saying where to find the+ Corresponding Source. Regardless of what server hosts the+ Corresponding Source, you remain obligated to ensure that it is+ available for as long as needed to satisfy these requirements.++ e) Convey the object code using peer-to-peer transmission, provided+ you inform other peers where the object code and Corresponding+ Source of the work are being offered to the general public at no+ charge under subsection 6d.++ A separable portion of the object code, whose source code is excluded+from the Corresponding Source as a System Library, need not be+included in conveying the object code work.++ A "User Product" is either (1) a "consumer product", which means any+tangible personal property which is normally used for personal, family,+or household purposes, or (2) anything designed or sold for incorporation+into a dwelling. In determining whether a product is a consumer product,+doubtful cases shall be resolved in favor of coverage. For a particular+product received by a particular user, "normally used" refers to a+typical or common use of that class of product, regardless of the status+of the particular user or of the way in which the particular user+actually uses, or expects or is expected to use, the product. A product+is a consumer product regardless of whether the product has substantial+commercial, industrial or non-consumer uses, unless such uses represent+the only significant mode of use of the product.++ "Installation Information" for a User Product means any methods,+procedures, authorization keys, or other information required to install+and execute modified versions of a covered work in that User Product from+a modified version of its Corresponding Source. The information must+suffice to ensure that the continued functioning of the modified object+code is in no case prevented or interfered with solely because+modification has been made.++ If you convey an object code work under this section in, or with, or+specifically for use in, a User Product, and the conveying occurs as+part of a transaction in which the right of possession and use of the+User Product is transferred to the recipient in perpetuity or for a+fixed term (regardless of how the transaction is characterized), the+Corresponding Source conveyed under this section must be accompanied+by the Installation Information. But this requirement does not apply+if neither you nor any third party retains the ability to install+modified object code on the User Product (for example, the work has+been installed in ROM).++ The requirement to provide Installation Information does not include a+requirement to continue to provide support service, warranty, or updates+for a work that has been modified or installed by the recipient, or for+the User Product in which it has been modified or installed. Access to a+network may be denied when the modification itself materially and+adversely affects the operation of the network or violates the rules and+protocols for communication across the network.++ Corresponding Source conveyed, and Installation Information provided,+in accord with this section must be in a format that is publicly+documented (and with an implementation available to the public in+source code form), and must require no special password or key for+unpacking, reading or copying.++ 7. Additional Terms.++ "Additional permissions" are terms that supplement the terms of this+License by making exceptions from one or more of its conditions.+Additional permissions that are applicable to the entire Program shall+be treated as though they were included in this License, to the extent+that they are valid under applicable law. If additional permissions+apply only to part of the Program, that part may be used separately+under those permissions, but the entire Program remains governed by+this License without regard to the additional permissions.++ When you convey a copy of a covered work, you may at your option+remove any additional permissions from that copy, or from any part of+it. (Additional permissions may be written to require their own+removal in certain cases when you modify the work.) You may place+additional permissions on material, added by you to a covered work,+for which you have or can give appropriate copyright permission.++ Notwithstanding any other provision of this License, for material you+add to a covered work, you may (if authorized by the copyright holders of+that material) supplement the terms of this License with terms:++ a) Disclaiming warranty or limiting liability differently from the+ terms of sections 15 and 16 of this License; or++ b) Requiring preservation of specified reasonable legal notices or+ author attributions in that material or in the Appropriate Legal+ Notices displayed by works containing it; or++ c) Prohibiting misrepresentation of the origin of that material, or+ requiring that modified versions of such material be marked in+ reasonable ways as different from the original version; or++ d) Limiting the use for publicity purposes of names of licensors or+ authors of the material; or++ e) Declining to grant rights under trademark law for use of some+ trade names, trademarks, or service marks; or++ f) Requiring indemnification of licensors and authors of that+ material by anyone who conveys the material (or modified versions of+ it) with contractual assumptions of liability to the recipient, for+ any liability that these contractual assumptions directly impose on+ those licensors and authors.++ All other non-permissive additional terms are considered "further+restrictions" within the meaning of section 10. If the Program as you+received it, or any part of it, contains a notice stating that it is+governed by this License along with a term that is a further+restriction, you may remove that term. If a license document contains+a further restriction but permits relicensing or conveying under this+License, you may add to a covered work material governed by the terms+of that license document, provided that the further restriction does+not survive such relicensing or conveying.++ If you add terms to a covered work in accord with this section, you+must place, in the relevant source files, a statement of the+additional terms that apply to those files, or a notice indicating+where to find the applicable terms.++ Additional terms, permissive or non-permissive, may be stated in the+form of a separately written license, or stated as exceptions;+the above requirements apply either way.++ 8. Termination.++ You may not propagate or modify a covered work except as expressly+provided under this License. Any attempt otherwise to propagate or+modify it is void, and will automatically terminate your rights under+this License (including any patent licenses granted under the third+paragraph of section 11).++ However, if you cease all violation of this License, then your+license from a particular copyright holder is reinstated (a)+provisionally, unless and until the copyright holder explicitly and+finally terminates your license, and (b) permanently, if the copyright+holder fails to notify you of the violation by some reasonable means+prior to 60 days after the cessation.++ Moreover, your license from a particular copyright holder is+reinstated permanently if the copyright holder notifies you of the+violation by some reasonable means, this is the first time you have+received notice of violation of this License (for any work) from that+copyright holder, and you cure the violation prior to 30 days after+your receipt of the notice.++ Termination of your rights under this section does not terminate the+licenses of parties who have received copies or rights from you under+this License. If your rights have been terminated and not permanently+reinstated, you do not qualify to receive new licenses for the same+material under section 10.++ 9. Acceptance Not Required for Having Copies.++ You are not required to accept this License in order to receive or+run a copy of the Program. Ancillary propagation of a covered work+occurring solely as a consequence of using peer-to-peer transmission+to receive a copy likewise does not require acceptance. However,+nothing other than this License grants you permission to propagate or+modify any covered work. These actions infringe copyright if you do+not accept this License. Therefore, by modifying or propagating a+covered work, you indicate your acceptance of this License to do so.++ 10. Automatic Licensing of Downstream Recipients.++ Each time you convey a covered work, the recipient automatically+receives a license from the original licensors, to run, modify and+propagate that work, subject to this License. You are not responsible+for enforcing compliance by third parties with this License.++ An "entity transaction" is a transaction transferring control of an+organization, or substantially all assets of one, or subdividing an+organization, or merging organizations. If propagation of a covered+work results from an entity transaction, each party to that+transaction who receives a copy of the work also receives whatever+licenses to the work the party's predecessor in interest had or could+give under the previous paragraph, plus a right to possession of the+Corresponding Source of the work from the predecessor in interest, if+the predecessor has it or can get it with reasonable efforts.++ You may not impose any further restrictions on the exercise of the+rights granted or affirmed under this License. For example, you may+not impose a license fee, royalty, or other charge for exercise of+rights granted under this License, and you may not initiate litigation+(including a cross-claim or counterclaim in a lawsuit) alleging that+any patent claim is infringed by making, using, selling, offering for+sale, or importing the Program or any portion of it.++ 11. Patents.++ A "contributor" is a copyright holder who authorizes use under this+License of the Program or a work on which the Program is based. The+work thus licensed is called the contributor's "contributor version".++ A contributor's "essential patent claims" are all patent claims+owned or controlled by the contributor, whether already acquired or+hereafter acquired, that would be infringed by some manner, permitted+by this License, of making, using, or selling its contributor version,+but do not include claims that would be infringed only as a+consequence of further modification of the contributor version. For+purposes of this definition, "control" includes the right to grant+patent sublicenses in a manner consistent with the requirements of+this License.++ Each contributor grants you a non-exclusive, worldwide, royalty-free+patent license under the contributor's essential patent claims, to+make, use, sell, offer for sale, import and otherwise run, modify and+propagate the contents of its contributor version.++ In the following three paragraphs, a "patent license" is any express+agreement or commitment, however denominated, not to enforce a patent+(such as an express permission to practice a patent or covenant not to+sue for patent infringement). To "grant" such a patent license to a+party means to make such an agreement or commitment not to enforce a+patent against the party.++ If you convey a covered work, knowingly relying on a patent license,+and the Corresponding Source of the work is not available for anyone+to copy, free of charge and under the terms of this License, through a+publicly available network server or other readily accessible means,+then you must either (1) cause the Corresponding Source to be so+available, or (2) arrange to deprive yourself of the benefit of the+patent license for this particular work, or (3) arrange, in a manner+consistent with the requirements of this License, to extend the patent+license to downstream recipients. "Knowingly relying" means you have+actual knowledge that, but for the patent license, your conveying the+covered work in a country, or your recipient's use of the covered work+in a country, would infringe one or more identifiable patents in that+country that you have reason to believe are valid.++ If, pursuant to or in connection with a single transaction or+arrangement, you convey, or propagate by procuring conveyance of, a+covered work, and grant a patent license to some of the parties+receiving the covered work authorizing them to use, propagate, modify+or convey a specific copy of the covered work, then the patent license+you grant is automatically extended to all recipients of the covered+work and works based on it.++ A patent license is "discriminatory" if it does not include within+the scope of its coverage, prohibits the exercise of, or is+conditioned on the non-exercise of one or more of the rights that are+specifically granted under this License. You may not convey a covered+work if you are a party to an arrangement with a third party that is+in the business of distributing software, under which you make payment+to the third party based on the extent of your activity of conveying+the work, and under which the third party grants, to any of the+parties who would receive the covered work from you, a discriminatory+patent license (a) in connection with copies of the covered work+conveyed by you (or copies made from those copies), or (b) primarily+for and in connection with specific products or compilations that+contain the covered work, unless you entered into that arrangement,+or that patent license was granted, prior to 28 March 2007.++ Nothing in this License shall be construed as excluding or limiting+any implied license or other defenses to infringement that may+otherwise be available to you under applicable patent law.++ 12. No Surrender of Others' Freedom.++ If conditions are imposed on you (whether by court order, agreement or+otherwise) that contradict the conditions of this License, they do not+excuse you from the conditions of this License. If you cannot convey a+covered work so as to satisfy simultaneously your obligations under this+License and any other pertinent obligations, then as a consequence you may+not convey it at all. For example, if you agree to terms that obligate you+to collect a royalty for further conveying from those to whom you convey+the Program, the only way you could satisfy both those terms and this+License would be to refrain entirely from conveying the Program.++ 13. Remote Network Interaction; Use with the GNU General Public License.++ Notwithstanding any other provision of this License, if you modify the+Program, your modified version must prominently offer all users+interacting with it remotely through a computer network (if your version+supports such interaction) an opportunity to receive the Corresponding+Source of your version by providing access to the Corresponding Source+from a network server at no charge, through some standard or customary+means of facilitating copying of software. This Corresponding Source+shall include the Corresponding Source for any work covered by version 3+of the GNU General Public License that is incorporated pursuant to the+following paragraph.++ Notwithstanding any other provision of this License, you have+permission to link or combine any covered work with a work licensed+under version 3 of the GNU General Public License into a single+combined work, and to convey the resulting work. The terms of this+License will continue to apply to the part which is the covered work,+but the work with which it is combined will remain governed by version+3 of the GNU General Public License.++ 14. Revised Versions of this License.++ The Free Software Foundation may publish revised and/or new versions of+the GNU Affero General Public License from time to time. Such new versions+will be similar in spirit to the present version, but may differ in detail to+address new problems or concerns.++ Each version is given a distinguishing version number. If the+Program specifies that a certain numbered version of the GNU Affero General+Public License "or any later version" applies to it, you have the+option of following the terms and conditions either of that numbered+version or of any later version published by the Free Software+Foundation. If the Program does not specify a version number of the+GNU Affero General Public License, you may choose any version ever published+by the Free Software Foundation.++ If the Program specifies that a proxy can decide which future+versions of the GNU Affero General Public License can be used, that proxy's+public statement of acceptance of a version permanently authorizes you+to choose that version for the Program.++ Later license versions may give you additional or different+permissions. However, no additional obligations are imposed on any+author or copyright holder as a result of your choosing to follow a+later version.++ 15. Disclaimer of Warranty.++ THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY+APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT+HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY+OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO,+THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR+PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM+IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF+ALL NECESSARY SERVICING, REPAIR OR CORRECTION.++ 16. Limitation of Liability.++ IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING+WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MODIFIES AND/OR CONVEYS+THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY+GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE+USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF+DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD+PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS),+EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF+SUCH DAMAGES.++ 17. Interpretation of Sections 15 and 16.++ If the disclaimer of warranty and limitation of liability provided+above cannot be given local legal effect according to their terms,+reviewing courts shall apply local law that most closely approximates+an absolute waiver of all civil liability in connection with the+Program, unless a warranty or assumption of liability accompanies a+copy of the Program in return for a fee.++ END OF TERMS AND CONDITIONS++ How to Apply These Terms to Your New Programs++ If you develop a new program, and you want it to be of the greatest+possible use to the public, the best way to achieve this is to make it+free software which everyone can redistribute and change under these terms.++ To do so, attach the following notices to the program. It is safest+to attach them to the start of each source file to most effectively+state the exclusion of warranty; and each file should have at least+the "copyright" line and a pointer to where the full notice is found.++ <one line to give the program's name and a brief idea of what it does.>+ Copyright (C) <year> <name of author>++ This program is free software: you can redistribute it and/or modify+ it under the terms of the GNU Affero General Public License as published by+ the Free Software Foundation, either version 3 of the License, or+ (at your option) any later version.++ This program is distributed in the hope that it will be useful,+ but WITHOUT ANY WARRANTY; without even the implied warranty of+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the+ GNU Affero General Public License for more details.++ You should have received a copy of the GNU Affero General Public License+ along with this program. If not, see <http://www.gnu.org/licenses/>.++Also add information on how to contact you by electronic and paper mail.++ If your software can interact with users remotely through a computer+network, you should also make sure that it provides a way for users to+get its source. For example, if your program is a web application, its+interface could display a "Source" link that leads users to an archive+of the code. There are many ways you could offer source, and different+solutions will be better for different programs; see section 13 for the+specific requirements.++ You should also get your employer (if you work as a programmer) or school,+if any, to sign a "copyright disclaimer" for the program, if necessary.+For more information on this, and how to apply and follow the GNU AGPL, see+<http://www.gnu.org/licenses/>.
+ README.md view
@@ -0,0 +1,39 @@+# Haskell PGMQ++This is a Haskell interface to Tembo's+[pgmq](https://github.com/tembo-io/pgmq) PostgreSQL extension.++The API design is based on [Elixir PGMQ](https://hexdocs.pm/pgmq/Pgmq.html) bindings.++## Running test cases++There is a binary located at [./bin/simple-test/Main.hs](./bin/simple-test/Main.hs), which contains all test cases+for the basic module `Database.PGMQ.Simple`.++First, choose which container image to use:+```shell+export PGMQ_CONTAINER=quay.io/tembo/pg16-pgmq:latest+```+However, with the above container, some tests might fail because certain functions+throw errors when a table is not present (e.g. `pgmq.drop_queue`).++Some fixes are available in our custom repository:+https://github.com/garganscript/pgmq+and you can use:+```shell+export PGMQ_CONTAINER=cgenie/pgmq:16-1.3.3.1+```++To run the tests, first start a PGMQ container:+```shell+ podman run --rm --name pgmq -e POSTGRES_PASSWORD=postgres -p 5432:5432 $PGMQ_CONTAINER+```+or with Docker:+```shell+ docker run --rm --name pgmq -e POSTGRES_PASSWORD=postgres -p 5432:5432 $PGMQ_CONTAINER+```+Then, execute the tests:+```shell+ cabal v2-test+```+The tests should print debug information and complete without errors.
+ bin/simple-exception-catch/Main.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE QuasiQuotes #-}++{-+Testing exception catch for PSQL.+-}+++module Main+where++import Control.Exception (SomeException(..), catch)+import Control.Monad (void)+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.SqlQQ (sql)+++main :: IO ()+main = do+ let connInfo = PSQL.defaultConnectInfo { PSQL.connectUser = "postgres"+ , PSQL.connectDatabase = "postgres" }+ conn <- PSQL.connect connInfo++ void (PSQL.execute_ conn [sql| CREATE EXTENSION IF NOT EXISTS pgmq |])++ let queue = "test" :: String+ -- catch exception of deleting a queue that doesn't exist+ catch (+ void (PSQL.query conn [sql| SELECT pgmq.drop_queue(?) |] (PSQL.Only queue) :: IO [PSQL.Only Bool])+ ) (\e@(SomeException _) -> putStrLn $ "err: " <> show e)++ return ()
+ bin/simple-test/Main.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE QuasiQuotes #-}++{-+Database.PGMQ.Simple functions tests.+-}+++module Main+where++import Control.Monad (void)+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PGMQ.Simple qualified as PGMQ+import Database.PGMQ.Types qualified as PGMQ+++main :: IO ()+main = do+ let connInfo = PSQL.defaultConnectInfo { PSQL.connectUser = "postgres"+ , PSQL.connectDatabase = "postgres" }+ conn <- PSQL.connect connInfo++ PGMQ.initialize conn++ PGMQ.dropQueue conn "test"+ -- _metrics <- PGMQ.getMetrics conn "test"++ let queue = "test"+ PGMQ.createQueue conn "test"+ void $ PGMQ.sendMessage conn queue ("test message 1" :: String) 0+ void $ PGMQ.sendMessage conn queue ("test message 2" :: String) 0++ allMetrics <- PGMQ.getMetricsAll conn+ putStrLn $ "allMetrics: " <> show allMetrics++ testMetrics <- PGMQ.getMetrics conn queue+ putStrLn $ "testMetrics: " <> show testMetrics++ + mMsg1 <- (PGMQ.readMessage conn queue 10 :: IO (Maybe (PGMQ.Message String)))+ case mMsg1 of+ Nothing -> undefined+ Just msg -> do+ putStrLn $ "message 1: " <> PGMQ.message msg+ + PGMQ.archiveMessage conn queue (PGMQ.msgId msg)+++ mMsg2 <- (PGMQ.readMessage conn queue 10 :: IO (Maybe (PGMQ.Message String)))+ case mMsg2 of+ Nothing -> undefined+ Just msg -> do+ putStrLn $ "message 2: " <> PGMQ.message msg+ + PGMQ.deleteMessage conn queue (PGMQ.msgId msg)++ + void $ PGMQ.sendMessages conn queue (["test message 3", "test message 4"] :: [String]) 0++ messagesA <- (PGMQ.readMessages conn queue 10 2 :: IO [PGMQ.Message String])+ PGMQ.archiveMessages conn queue [PGMQ.msgId msg | msg <- messagesA]+++ void $ PGMQ.sendMessages conn queue (["test message 5", "test message 6"] :: [String]) 0++ messagesD <- (PGMQ.readMessages conn queue 10 2 :: IO [PGMQ.Message String])+ PGMQ.deleteMessages conn queue [PGMQ.msgId msg | msg <- messagesD]++ queues <- PGMQ.listQueues conn+ putStrLn $ "queues: " <> show queues+ ++ void $ PGMQ.sendMessage conn queue ("test message 7" :: String) 0+ mMsgP <- (PGMQ.popMessage conn queue :: IO (Maybe (PGMQ.Message String)))+ case mMsgP of+ Nothing -> undefined+ Just msg -> do+ putStrLn $ "message pop: " <> PGMQ.message msg+ PGMQ.setMessageVt conn queue (PGMQ.msgId msg) 10++ void $ PGMQ.sendMessage conn queue ("test message 8" :: String) 0+ metrics <- PGMQ.getMetrics conn queue+ putStrLn $ "before purge: " <> show metrics+ PGMQ.purgeQueue conn queue+ metrics2 <- PGMQ.getMetrics conn queue+ putStrLn $ "after purge: " <> show metrics2+ + -- PGMQ.dropQueue conn queue+ + return ()
+ haskell-pgmq.cabal view
@@ -0,0 +1,205 @@+cabal-version: 3.4+-- The cabal-version field refers to the version of the .cabal specification,+-- and can be different from the cabal-install (the tool) version and the+-- Cabal (the library) version you are using. As such, the Cabal (the library)+-- version used must be equal or greater than the version stated in this field.+-- Starting from the specification version 2.2, the cabal-version field must be+-- the first thing in the cabal file.++-- Initial package description 'haskell-pgmq' generated by+-- 'cabal init'. For further documentation, see:+-- http://haskell.org/cabal/users-guide/+--+-- The name of the package.+name: haskell-pgmq++-- The package version.+-- See the Haskell package versioning policy (PVP) for standards+-- guiding when and how versions should be incremented.+-- https://pvp.haskell.org+-- PVP summary: +-+------- breaking API changes+-- | | +----- non-breaking API additions+-- | | | +--- code changes with no API change+version: 0.1.0.0++-- A short (one-line) description of the package.+synopsis: Haskell interface for Tembo's PGMQ PostgreSQL extension+ +-- Extra doc files to be distributed with the package, such as a CHANGELOG or a README.+extra-doc-files: README.md++-- A longer description of the package.+description: + Please see the README on GitHub at <https://gitlab.iscpif.fr/gargantext/haskell-pgmq/blob/master/README.md>++-- URL for the project homepage or repository.+homepage: https://gitlab.iscpif.fr/gargantext/haskell-pgmq++-- The license under which the package is released.+license: AGPL-3.0-or-later++-- The file containing the license text.+license-file: LICENSE++-- The package author(s).+author: Gargantext++-- An email address to which users can send suggestions, bug reports, and patches.+maintainer: gargantext@iscpif.fr+ +-- A copyright notice.+-- copyright:+category: Database+build-type: Simple++-- Extra doc files to be distributed with the package, such as a CHANGELOG or a README.+extra-doc-files: CHANGELOG.md++-- Extra source files to be distributed with the package, such as examples, or a tutorial module.+-- extra-source-files:++data-files:+ sql/pgmq.sql++common warnings+ ghc-options: -Wall++source-repository head+ type: git+ location: https://gitlab.iscpif.fr/gargantext/haskell-pgmq.git++library+ -- Import common warning flags.+ import: warnings++ -- Modules exported by the library.+ exposed-modules: Database.PGMQ+ , Database.PGMQ.Helpers+ , Database.PGMQ.Schema+ , Database.PGMQ.Simple+ , Database.PGMQ.Types+ + other-modules: Paths_haskell_pgmq++ autogen-modules: Paths_haskell_pgmq++ -- Modules included in this library but not exported.+ -- other-modules:++ -- LANGUAGE extensions used by modules in this package.+ -- other-extensions:++ -- Other library packages from which modules are imported.+ build-depends: base >=4.17.2.0 && < 5+ , aeson >= 1.4 && < 2.3+ , bytestring >= 0.11 && < 0.13+ , postgresql-simple >= 0.6 && < 0.8+ , safe >= 0.3 && < 0.4+ , text >= 1.2 && < 2.2+ , time >= 1.10 && < 1.15+ , units >= 2.4 && < 2.5++ -- Directories containing source files.+ hs-source-dirs: src++ -- Base language which the package is written in.+ default-language: Haskell2010++ default-extensions:+ DuplicateRecordFields+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ NamedFieldPuns+ OverloadedStrings+ RecordWildCards++ +executable simple-exception-catch+ -- Import common warning flags.+ import: warnings++ build-depends: base >=4.17.2.0 && < 5+ , postgresql-simple >= 0.6 && < 0.8+ , haskell-pgmq++ -- Directories containing source files.+ hs-source-dirs: bin/simple-exception-catch+ main-is: Main.hs++ -- Base language which the package is written in.+ default-language: Haskell2010++ default-extensions:+ DuplicateRecordFields+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ NamedFieldPuns+ OverloadedStrings+ RecordWildCards++ ghc-options: -threaded++executable simple-test+ -- Import common warning flags.+ import: warnings++ build-depends: base >=4.17.2.0 && < 5+ , postgresql-simple >= 0.6 && < 0.8+ , haskell-pgmq++ -- Directories containing source files.+ hs-source-dirs: bin/simple-test+ main-is: Main.hs++ -- Base language which the package is written in.+ default-language: Haskell2010++ default-extensions:+ DuplicateRecordFields+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ NamedFieldPuns+ OverloadedStrings+ RecordWildCards++ ghc-options: -threaded++test-suite test-integration+ -- Import common warning flags.+ import: warnings+ + type: exitcode-stdio-1.0++ other-modules: Test.Integration.PGMQ.Helpers+ , Test.Integration.PGMQ.Simple+ , Test.Integration.Utils+ + build-depends: base >=4.17.2.0 && < 5+ , aeson >= 1.4 && < 2.3+ , containers >= 0.6 && < 0.8+ , hspec >= 2.11 && < 3+ , postgresql-simple >= 0.6 && < 0.8+ , random-strings == 0.1.1.0+ , stm >= 2.5.3 && < 3+ , tasty >= 1.5 && < 1.6+ , tasty-hspec >= 1.2.0 && < 2+ , text >= 1.2 && < 2.2+ + , haskell-pgmq++ -- Directories containing source files.+ hs-source-dirs: tests+ main-is: integration-tests.hs++ -- Base language which the package is written in.+ default-language: Haskell2010++ default-extensions:+ DuplicateRecordFields+ GeneralizedNewtypeDeriving+ ImportQualifiedPost+ NamedFieldPuns+ OverloadedStrings+ RecordWildCards++ ghc-options: -threaded
+ sql/pgmq.sql view
@@ -0,0 +1,778 @@+-- Copied from+-- git@github.com:garganscript/pgmq.git+-- at rev+-- e2740d0da8e663618feb19e9c6f6653de861fe14+------------------------------------------------------------+-- Schema, tables, records, privileges, indexes, etc+------------------------------------------------------------+-- We don't need to create the `pgmq` schema because it is automatically+-- created by postgres due to being declared in extension control file++-- Table where queues and metadata about them is stored+CREATE TABLE pgmq.meta (+ queue_name VARCHAR UNIQUE NOT NULL,+ is_partitioned BOOLEAN NOT NULL,+ is_unlogged BOOLEAN NOT NULL,+ created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL+);++-- Grant permission to pg_monitor to all tables and sequences+GRANT USAGE ON SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor;++-- This type has the shape of a message in a queue, and is often returned by+-- pgmq functions that return messages+CREATE TYPE pgmq.message_record AS (+ msg_id BIGINT,+ read_ct INTEGER,+ enqueued_at TIMESTAMP WITH TIME ZONE,+ vt TIMESTAMP WITH TIME ZONE,+ message JSONB+);++CREATE TYPE pgmq.queue_record AS (+ queue_name VARCHAR,+ is_partitioned BOOLEAN,+ is_unlogged BOOLEAN,+ created_at TIMESTAMP WITH TIME ZONE+);++------------------------------------------------------------+-- Functions+------------------------------------------------------------+-- read+-- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.q_%s+ WHERE vt <= clock_timestamp()+ ORDER BY msg_id ASC+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.q_%s m+ SET+ vt = clock_timestamp() + interval '%s seconds',+ read_ct = read_ct + 1+ FROM cte+ WHERE m.msg_id = cte.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.vt, m.message;+ $QUERY$,+ queue_name, queue_name, vt+ );+ RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++---- read_with_poll+---- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read_with_poll(+ queue_name TEXT,+ vt INTEGER,+ qty INTEGER,+ max_poll_seconds INTEGER DEFAULT 5,+ poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ r pgmq.message_record;+ stop_at TIMESTAMP;+ sql TEXT;+BEGIN+ stop_at := clock_timestamp() + FORMAT('%s seconds', max_poll_seconds)::interval;+ LOOP+ IF (SELECT clock_timestamp() >= stop_at) THEN+ RETURN;+ END IF;++ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.q_%s+ WHERE vt <= clock_timestamp()+ ORDER BY msg_id ASC+ LIMIT $1+ FOR UPDATE SKIP LOCKED+ )+ UPDATE pgmq.q_%s m+ SET+ vt = clock_timestamp() + interval '%s seconds',+ read_ct = read_ct + 1+ FROM cte+ WHERE m.msg_id = cte.msg_id+ RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.vt, m.message;+ $QUERY$,+ queue_name, queue_name, vt+ );++ FOR r IN+ EXECUTE sql USING qty+ LOOP+ RETURN NEXT r;+ END LOOP;+ IF FOUND THEN+ RETURN;+ ELSE+ PERFORM pg_sleep(poll_interval_ms / 1000);+ END IF;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes a message from the queue, and sends it to the archive, where its+---- saved permanently.+CREATE FUNCTION pgmq.archive(+ queue_name TEXT,+ msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BIGINT;+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH archived AS (+ DELETE FROM pgmq.q_%s+ WHERE msg_id = $1+ RETURNING msg_id, vt, read_ct, enqueued_at, message+ )+ INSERT INTO pgmq.a_%s (msg_id, vt, read_ct, enqueued_at, message)+ SELECT msg_id, vt, read_ct, enqueued_at, message+ FROM archived+ RETURNING msg_id;+ $QUERY$,+ queue_name, queue_name+ );+ EXECUTE sql USING msg_id INTO result;+ RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes an array of message ids from the queue, and sends it to the archive,+---- where these messages will be saved permanently.+CREATE FUNCTION pgmq.archive(+ queue_name TEXT,+ msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH archived AS (+ DELETE FROM pgmq.q_%s+ WHERE msg_id = ANY($1)+ RETURNING msg_id, vt, read_ct, enqueued_at, message+ )+ INSERT INTO pgmq.a_%s (msg_id, vt, read_ct, enqueued_at, message)+ SELECT msg_id, vt, read_ct, enqueued_at, message+ FROM archived+ RETURNING msg_id;+ $QUERY$,+ queue_name, queue_name+ );+ RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes a message id from the queue permanently+CREATE FUNCTION pgmq.delete(+ queue_name TEXT,+ msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BIGINT;+BEGIN+ sql := FORMAT(+ $QUERY$+ DELETE FROM pgmq.q_%s+ WHERE msg_id = $1+ RETURNING msg_id+ $QUERY$,+ queue_name+ );+ EXECUTE sql USING msg_id INTO result;+ RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes an array of message ids from the queue permanently+CREATE FUNCTION pgmq.delete(+ queue_name TEXT,+ msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+BEGIN+ sql := FORMAT(+ $QUERY$+ DELETE FROM pgmq.q_%s+ WHERE msg_id = ANY($1)+ RETURNING msg_id+ $QUERY$,+ queue_name+ );+ RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++-- send+-- sends a message to a queue, optionally with a delay+CREATE FUNCTION pgmq.send(+ queue_name TEXT,+ msg JSONB,+ delay INTEGER DEFAULT 0+) RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+BEGIN+ sql := FORMAT(+ $QUERY$+ INSERT INTO pgmq.q_%s (vt, message)+ VALUES ((clock_timestamp() + interval '%s seconds'), $1)+ RETURNING msg_id;+ $QUERY$,+ queue_name, delay+ );+ RETURN QUERY EXECUTE sql USING msg;+END;+$$ LANGUAGE plpgsql;++-- send_batch+-- sends an array of list of messages to a queue, optionally with a delay+CREATE FUNCTION pgmq.send_batch(+ queue_name TEXT,+ msgs JSONB[],+ delay INTEGER DEFAULT 0+) RETURNS SETOF BIGINT AS $$+DECLARE+ sql TEXT;+BEGIN+ sql := FORMAT(+ $QUERY$+ INSERT INTO pgmq.q_%s (vt, message)+ SELECT clock_timestamp() + interval '%s seconds', unnest($1)+ RETURNING msg_id;+ $QUERY$,+ queue_name, delay+ );+ RETURN QUERY EXECUTE sql USING msgs;+END;+$$ LANGUAGE plpgsql;++-- returned by pgmq.metrics() and pgmq.metrics_all+CREATE TYPE pgmq.metrics_result AS (+ queue_name text,+ queue_length bigint,+ newest_msg_age_sec int,+ oldest_msg_age_sec int,+ total_messages bigint,+ scrape_time timestamp with time zone+);++-- get metrics for a single queue+CREATE FUNCTION pgmq.metrics(queue_name TEXT)+RETURNS pgmq.metrics_result AS $$+DECLARE+ result_row pgmq.metrics_result;+ query TEXT;+BEGIN+ query := FORMAT(+ $QUERY$+ WITH q_summary AS (+ SELECT+ count(*) as queue_length,+ EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,+ EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,+ NOW() as scrape_time+ FROM pgmq.q_%s+ ),+ all_metrics AS (+ SELECT CASE+ WHEN is_called THEN last_value ELSE 0+ END as total_messages+ FROM pgmq.q_%s_msg_id_seq+ )+ SELECT+ '%s' as queue_name,+ q_summary.queue_length,+ q_summary.newest_msg_age_sec,+ q_summary.oldest_msg_age_sec,+ all_metrics.total_messages,+ q_summary.scrape_time+ FROM q_summary, all_metrics+ $QUERY$,+ queue_name, queue_name, queue_name+ );+ EXECUTE query INTO result_row;+ RETURN result_row;+END;+$$ LANGUAGE plpgsql;++-- get metrics for all queues+CREATE FUNCTION pgmq."metrics_all"()+RETURNS SETOF pgmq.metrics_result AS $$+DECLARE+ row_name RECORD;+ result_row pgmq.metrics_result;+BEGIN+ FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP+ result_row := pgmq.metrics(row_name.queue_name);+ RETURN NEXT result_row;+ END LOOP;+END;+$$ LANGUAGE plpgsql;++-- list queues+CREATE FUNCTION pgmq."list_queues"()+RETURNS SETOF pgmq.queue_record AS $$+BEGIN+ RETURN QUERY SELECT * FROM pgmq.meta;+END+$$ LANGUAGE plpgsql;++-- purge queue, deleting all entries in it.+CREATE FUNCTION pgmq."purge_queue"(queue_name TEXT)+RETURNS BIGINT AS $$+DECLARE+ deleted_count INTEGER;+BEGIN+ EXECUTE format('DELETE FROM pgmq.q_%s', queue_name);+ GET DIAGNOSTICS deleted_count = ROW_COUNT;+ RETURN deleted_count;+END+$$ LANGUAGE plpgsql;++-- unassign archive, so it can be kept when a queue is deleted+CREATE FUNCTION pgmq."detach_archive"(queue_name TEXT)+RETURNS VOID AS $$+BEGIN+ EXECUTE format('ALTER EXTENSION pgmq DROP TABLE pgmq.a_%s', queue_name);+END+$$ LANGUAGE plpgsql;++-- pop a single message+CREATE FUNCTION pgmq.pop(queue_name TEXT)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ result pgmq.message_record;+BEGIN+ sql := FORMAT(+ $QUERY$+ WITH cte AS+ (+ SELECT msg_id+ FROM pgmq.q_%s+ WHERE vt <= now()+ ORDER BY msg_id ASC+ LIMIT 1+ FOR UPDATE SKIP LOCKED+ )+ DELETE from pgmq.q_%s+ WHERE msg_id = (select msg_id from cte)+ RETURNING *;+ $QUERY$,+ queue_name, queue_name+ );+ RETURN QUERY EXECUTE sql;+END;+$$ LANGUAGE plpgsql;++-- Sets vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt INTEGER)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+ sql TEXT;+ result pgmq.message_record;+BEGIN+ sql := FORMAT(+ $QUERY$+ UPDATE pgmq.q_%s+ SET vt = (now() + interval '%s seconds')+ WHERE msg_id = %s+ RETURNING *;+ $QUERY$,+ queue_name, vt, msg_id+ );+ RETURN QUERY EXECUTE sql;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN DEFAULT FALSE)+RETURNS BOOLEAN AS $$+BEGIN+-- EXECUTE FORMAT(+-- $QUERY$+-- ALTER EXTENSION pgmq DROP TABLE pgmq.q_%s+-- $QUERY$,+-- queue_name+-- );+--+-- EXECUTE FORMAT(+-- $QUERY$+-- ALTER EXTENSION pgmq DROP TABLE pgmq.a_%s+-- $QUERY$,+-- queue_name+-- );++ EXECUTE FORMAT(+ $QUERY$+ DROP TABLE IF EXISTS pgmq.q_%s+ $QUERY$,+ queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ DROP TABLE IF EXISTS pgmq.a_%s+ $QUERY$,+ queue_name+ );++ IF EXISTS (+ SELECT 1+ FROM information_schema.tables+ WHERE table_name = 'meta' and table_schema = 'pgmq'+ ) THEN+ EXECUTE FORMAT(+ $QUERY$+ DELETE FROM pgmq.meta WHERE queue_name = '%s'+ $QUERY$,+ queue_name+ );+ END IF;++ IF partitioned THEN+ EXECUTE FORMAT(+ $QUERY$+ DELETE FROM public.part_config where parent_table = '%s'+ $QUERY$,+ queue_name+ );+ END IF;++ RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.validate_queue_name(queue_name TEXT)+RETURNS void AS $$+BEGIN+ IF length(queue_name) >= 48 THEN+ RAISE EXCEPTION 'queue name is too long, maximum length is 48 characters';+ END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._belongs_to_pgmq(table_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+ sql TEXT;+ result BOOLEAN;+BEGIN+ SELECT EXISTS (+ SELECT 1+ FROM pg_depend+ WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq')+ AND objid = (+ SELECT oid+ FROM pg_class+ WHERE relname = table_name+ )+ ) INTO result;+ RETURN result;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_non_partitioned(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.q_%s (+ msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ )+ $QUERY$,+ queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.a_%s (+ msg_id BIGINT PRIMARY KEY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ );+ $QUERY$,+ queue_name+ );++-- IF NOT pgmq._belongs_to_pgmq(FORMAT('q_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.q_%s', queue_name);+-- END IF;+--+-- IF NOT pgmq._belongs_to_pgmq(FORMAT('a_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.a_%s', queue_name);+-- END IF;++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS q_%s_vt_idx ON pgmq.q_%s (vt ASC);+ $QUERY$,+ queue_name, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS archived_at_idx_%s ON pgmq.a_%s (archived_at);+ $QUERY$,+ queue_name, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES ('%s', false, false)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_unlogged(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);++ EXECUTE FORMAT(+ $QUERY$+ CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.q_%s (+ msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ )+ $QUERY$,+ queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.a_%s (+ msg_id BIGINT PRIMARY KEY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ );+ $QUERY$,+ queue_name+ );++-- IF NOT pgmq._belongs_to_pgmq(FORMAT('q_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.q_%s', queue_name);+-- END IF;+--+-- IF NOT pgmq._belongs_to_pgmq(FORMAT('a_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.a_%s', queue_name);+-- END IF;++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS q_%s_vt_idx ON pgmq.q_%s (vt ASC);+ $QUERY$,+ queue_name, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS archived_at_idx_%s ON pgmq.a_%s (archived_at);+ $QUERY$,+ queue_name, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES ('%s', false, true)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_partition_col(partition_interval TEXT)+RETURNS TEXT AS $$+DECLARE+ num INTEGER;+BEGIN+ BEGIN+ num := partition_interval::INTEGER;+ RETURN 'msg_id';+ EXCEPTION+ WHEN others THEN+ RETURN 'enqueued_at';+ END;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._ensure_pg_partman_installed()+RETURNS void AS $$+DECLARE+ extension_exists BOOLEAN;+BEGIN+ SELECT EXISTS (+ SELECT 1+ FROM pg_extension+ WHERE extname = 'pg_partman'+ ) INTO extension_exists;++ IF NOT extension_exists THEN+ RAISE EXCEPTION 'pg_partman is required for partitioned queues';+ END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_partitioned(+ queue_name TEXT,+ partition_interval TEXT DEFAULT '10000',+ retention_interval TEXT DEFAULT '100000'+)+RETURNS void AS $$+DECLARE+ partition_col TEXT;+BEGIN+ PERFORM pgmq.validate_queue_name(queue_name);+ PERFORM pgmq._ensure_pg_partman_installed();+ SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.q_%s (+ msg_id BIGINT GENERATED ALWAYS AS IDENTITY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ ) PARTITION BY RANGE (%s)+ $QUERY$,+ queue_name, partition_col+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE TABLE IF NOT EXISTS pgmq.a_%s (+ msg_id BIGINT PRIMARY KEY,+ read_ct INT DEFAULT 0 NOT NULL,+ enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+ vt TIMESTAMP WITH TIME ZONE NOT NULL,+ message JSONB+ );+ $QUERY$,+ queue_name+ );++-- IF NOT pgmq._belongs_to_pgmq(FORMAT('q_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.q_%s', queue_name);+-- END IF;+--+-- IF NOT pgmq._belongs_to_pgmq(FORMAT('a_%s', queue_name)) THEN+-- EXECUTE FORMAT('ALTER EXTENSION pgmq ADD TABLE pgmq.a_%s', queue_name);+-- END IF;++ EXECUTE FORMAT(+ $QUERY$+ SELECT public.create_parent('pgmq.q_%s', '%s', 'native', '%s');+ $QUERY$,+ queue_name, partition_col, partition_interval+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS q_%s_part_idx ON pgmq.q_%s (%s);+ $QUERY$,+ queue_name, queue_name, partition_col+ );++ EXECUTE FORMAT(+ $QUERY$+ CREATE INDEX IF NOT EXISTS archived_at_idx_%s ON pgmq.a_%s (archived_at);+ $QUERY$,+ queue_name, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ UPDATE public.part_config+ SET+ retention = '%s',+ retention_keep_table = false,+ retention_keep_index = true,+ automatic_maintenance = 'on'+ WHERE parent_table = 'pgmq.q_%s';+ $QUERY$,+ retention_interval, queue_name+ );++ EXECUTE FORMAT(+ $QUERY$+ INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+ VALUES ('%s', true, false)+ ON CONFLICT+ DO NOTHING;+ $QUERY$,+ queue_name+ );+END;+$$ LANGUAGE plpgsql;+++CREATE FUNCTION pgmq.create(queue_name TEXT)+RETURNS void AS $$+BEGIN+ PERFORM pgmq.create_non_partitioned(queue_name);+END;+$$ LANGUAGE plpgsql;
+ src/Database/PGMQ.hs view
@@ -0,0 +1,20 @@+{-|+Module : Database.PGMQ+Description : PGMQ Haskell bindings+Copyright : (c) Gargantext, 2024-Present+License : AGPL+Maintainer : gargantext@iscpif.fr+Stability : experimental+Portability : POSIX+-}++module Database.PGMQ+ ( module Database.PGMQ.Helpers+ , module Database.PGMQ.Simple+ , module Database.PGMQ.Types )+where++ +import Database.PGMQ.Helpers+import Database.PGMQ.Simple+import Database.PGMQ.Types
+ src/Database/PGMQ/Helpers.hs view
@@ -0,0 +1,34 @@+{-|+Module : Database.PGMQ.Helper+Description : Helper utilities around Database.PGMQ.Simple+Copyright : (c) Gargantext, 2024-Present+License : AGPL+Maintainer : gargantext@iscpif.fr+Stability : experimental+Portability : POSIX++-}++{-# LANGUAGE QuasiQuotes #-}+++module Database.PGMQ.Helpers+ ( safeSendMessage )+where+++import Database.PGMQ.Simple ( createQueue, sendMessage )+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.SqlQQ (sql)+import Database.PGMQ.Types (Delay, SerializableMessage, Queue)+++-- | This is a wrapper around 'sendMessage' that makes sure the queue exists.+safeSendMessage :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> a -> Delay -> IO Int+safeSendMessage conn queue msg delay = do+ PSQL.withTransaction conn $ do+ let magicLockId = 112233445566 :: Int+ _ <- PSQL.query conn [sql| SELECT pg_advisory_xact_lock(?) |] (PSQL.Only magicLockId) :: IO [PSQL.Only ()]+ createQueue conn queue+ sendMessage conn queue msg delay
+ src/Database/PGMQ/Schema.hs view
@@ -0,0 +1,9 @@+module Database.PGMQ.Schema where++import Paths_haskell_pgmq+++pgmqSchema :: IO String+pgmqSchema = do+ f <- getDataFileName "sql/pgmq.sql"+ readFile f
+ src/Database/PGMQ/Simple.hs view
@@ -0,0 +1,311 @@+{-|+Module : Database.PGMQ.Simple+Description : Basic functionality+Copyright : (c) Gargantext, 2024-Present+License : AGPL+Maintainer : gargantext@iscpif.fr+Stability : experimental+Portability : POSIX++https://tembo.io/pgmq/api/sql/functions/+-}++{-# LANGUAGE QuasiQuotes #-}++module Database.PGMQ.Simple+ ( initialize+ , archiveMessage+ , archiveMessages+ , createQueue+ , deleteMessage+ , deleteMessages+ , dropQueue+ , getMetrics+ , getMetricsAll+ , listQueues+ , popMessage+ , purgeQueue+ , readMessage+ , readMessageFromArchive+ , readMessageById+ , readMessages+ , readMessageWithPoll+ , readMessagesWithPoll+ , sendMessage+ , sendMessages+ , setMessageVt+ , queueAvailableLength+ , queueAvailableIds )+where+++-- Please note that message format should be JSON-serializable (the+-- 'pgmq.send' functions accept only a ::jsonb type).+++-- import Control.Exception (Exception, SomeException(..), catch, fromException, throwIO, toException)+import Control.Monad (void)+import Data.Functor ((<&>))+import Data.String (fromString)+import Data.Text qualified as T+import Database.PGMQ.Schema (pgmqSchema)+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.Newtypes qualified as PSQL (Aeson(..))+import Database.PostgreSQL.Simple.SqlQQ (sql)+import Database.PostgreSQL.Simple.Types qualified as PSQL (QualifiedIdentifier(..), PGArray(..))+import Database.PGMQ.Types (Delay, MaxPollSeconds, Message, MessageCount, MessageId, Metrics, SerializableMessage, PollIntervalMs, Queue, QueueInfo, VisibilityTimeout)+import Safe (headMay)+++{-| Initialize PGMQ given a PostgreSQL connection. Mainly concerned+ with creating the 'pgmq' extension. -}+initialize :: PSQL.Connection -> IO ()+initialize conn = do+ -- OK so this is a bit tricky because of the usage of IF NOT EXISTS:+ -- https://stackoverflow.com/questions/29900845/create-schema-if-not-exists-raises-duplicate-key-error+ -- PostgreSQL will complain badly if 'initialize' is called+ -- from multiple threads at once.+ -- Hence, we use 'pg_advisory_xact_lock' to lock ourselves+ -- out of this situation.+ pgmqSchema' <- pgmqSchema+ PSQL.withTransaction conn $ do+ let magicLockId = 1122334455 :: Int+ _ <- PSQL.query conn [sql| SELECT pg_advisory_xact_lock(?) |] (PSQL.Only magicLockId) :: IO [PSQL.Only ()]+ hasPgmq <- PSQL.query_ conn [sql|SELECT 1 FROM information_schema.schemata WHERE schema_name = 'pgmq'|] :: IO [PSQL.Only Int]+ case hasPgmq of+ [PSQL.Only 1] -> pure ()+ _ -> do+ -- NOTE This isn't exactly right because we initialize the+ -- extension by only running the SQL, not using `CREATE+ -- EXTENSION`. But it works and is quite hassle-free+ void $ PSQL.execute_ conn [sql|CREATE SCHEMA pgmq |]+ void $ PSQL.execute_ conn (fromString pgmqSchema')+ -- void $ PSQL.execute_ conn [sql| CREATE EXTENSION IF NOT EXISTS pgmq |]+++{-| Archives message in given queue for given id++ https://tembo.io/pgmq/api/sql/functions/#archive-single -}+archiveMessage :: PSQL.Connection -> Queue -> MessageId -> IO ()+archiveMessage conn queue msgId =+ void (PSQL.query conn [sql| SELECT pgmq.archive(?, ?) |] (queue, msgId) :: IO [PSQL.Only Bool])++{-| Archives messages in given queue for given ids++ https://tembo.io/pgmq/api/sql/functions/#archive-batch -}+archiveMessages :: PSQL.Connection -> Queue -> [MessageId] -> IO ()+archiveMessages conn queue msgIds =+ void (PSQL.query conn [sql| SELECT pgmq.archive(?, ?::bigint[]) |] (queue, PSQL.PGArray msgIds) :: IO [PSQL.Only Int])++{-| Creates a queue++ https://tembo.io/pgmq/api/sql/functions/#create -}+createQueue :: PSQL.Connection -> Queue -> IO ()+createQueue conn queue =+ void (PSQL.query conn [sql| SELECT pgmq.create(?) |] (PSQL.Only queue) :: IO [PSQL.Only ()])++{-| Deletes given message from given queue++ https://tembo.io/pgmq/api/sql/functions/#delete-single -}+deleteMessage :: PSQL.Connection -> Queue -> MessageId -> IO ()+deleteMessage conn queue msgId =+ void (PSQL.query conn [sql| SELECT pgmq.delete(?, ?) |] (queue, msgId) :: IO [PSQL.Only Bool])++{-| Deletes given messages from given queue++ https://tembo.io/pgmq/api/sql/functions/#delete-batch -}+deleteMessages :: PSQL.Connection -> Queue -> [MessageId] -> IO ()+deleteMessages conn queue msgIds =+ void (PSQL.query conn [sql| SELECT pgmq.delete(?, ?) |] (queue, PSQL.PGArray msgIds) :: IO [PSQL.Only Int])++{-| Deletes given queue++ https://tembo.io/pgmq/api/sql/functions/#drop_queue -}+dropQueue :: PSQL.Connection -> Queue -> IO ()+dropQueue conn queue =+ void (PSQL.query conn [sql| SELECT pgmq.drop_queue(?) |] (PSQL.Only queue) :: IO [PSQL.Only Bool])++{-| Read metrics for a given queue++ https://tembo.io/pgmq/api/sql/functions/#metrics -}+getMetrics :: PSQL.Connection -> Queue -> IO (Maybe Metrics)+getMetrics conn queue =+ PSQL.query conn [sql| SELECT * FROM pgmq.metrics(?) |] (PSQL.Only queue) <&> headMay+ -- catch+ -- (PSQL.query conn [sql| SELECT * FROM pgmq.metrics(?) |] (PSQL.Only queue) >>= return . headMay)+ -- handleError+ -- where+ -- -- support the case when a table does not exist+ -- -- handleError :: Exception e => e -> IO (Maybe Metrics)+ -- handleError err@(SomeException _) = do+ -- putStrLn $ "Error: " <> show err+ -- return Nothing+ -- handleError :: SomeException -> IO (Maybe Metrics)+ -- handleError err = do+ -- putStrLn $ "Error: " <> show err+ -- putStrLn "x"+ -- return Nothing+ -- case fromException err of+ -- -- Just (PSQL.SomePostgreSqlException e) ->+ -- -- case fromException (toException e) of+ -- -- Just (PSQL.SqlError { sqlState = "42P01" }) -> return Nothing+ -- -- -- re-raise other errors+ -- -- _ -> throwIO err+ -- Just (PSQL.SqlError { sqlState = "42P01" }) -> return Nothing+ -- -- re-raise other errors+ -- _ -> throwIO err+++-- | Read metrics for all queues+-- https://tembo.io/pgmq/api/sql/functions/#metrics_all+getMetricsAll :: PSQL.Connection -> IO [Metrics]+getMetricsAll conn =+ PSQL.query_ conn [sql| SELECT * FROM pgmq.metrics_all() |]++-- | List all pgmq queues+-- https://tembo.io/pgmq/api/sql/functions/#list_queues+listQueues :: PSQL.Connection -> IO [QueueInfo]+listQueues conn =+ PSQL.query_ conn [sql| SELECT * FROM pgmq.list_queues() |]++-- | Read a message and immediately delete it from the queue. Returns `None` if the queue is empty.+-- https://tembo.io/pgmq/api/sql/functions/#pop+popMessage :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> IO (Maybe (Message a))+popMessage conn queue = do+ ret <- PSQL.query conn+ [sql| SELECT msg_id, read_ct, enqueued_at, NULL, vt, message FROM pgmq.pop(?) |]+ (PSQL.Only queue)++ return $ headMay ret++-- | Deletes all messages from a queue+-- https://tembo.io/pgmq/api/sql/functions/#purge_queue+purgeQueue :: PSQL.Connection -> Queue -> IO ()+purgeQueue conn queue = do+ void (PSQL.query conn [sql| SELECT * FROM pgmq.purge_queue(?) |] (PSQL.Only queue) :: IO [PSQL.Only Int])++-- | Read a message from given queue, with given visibility timeout (in seconds)+-- https://tembo.io/pgmq/api/sql/functions/#read+readMessage :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> VisibilityTimeout -> IO (Maybe (Message a))+readMessage conn queue vt =+ readMessages conn queue vt 1 <&> headMay++-- | Read a message from given archive+readMessageFromArchive :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> MessageId -> IO (Maybe (Message a))+readMessageFromArchive conn queue msgId = do+ let archiveTable = "a_" <> queue+ PSQL.query conn+ [sql| SELECT msg_id, read_ct, enqueued_at, archived_at, vt, message FROM ? WHERE msg_id = ?|]+ (PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack archiveTable, msgId) <&> headMay++-- | Read a message from queue with given ID (for querying purposes, doesn't pop message from queue)+readMessageById :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> MessageId -> IO (Maybe (Message a))+readMessageById conn queue msgId = do+ let queueTable = "q_" <> queue+ PSQL.query conn+ [sql| SELECT msg_id, read_ct, enqueued_at, NULL, vt, message FROM ? WHERE msg_id = ?|]+ (PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack queueTable, msgId) <&> headMay++{-| Reads given number of messages from given queue++ https://tembo.io/pgmq/api/sql/functions/#read -}+readMessages :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> VisibilityTimeout -> MessageCount -> IO [Message a]+readMessages conn queue vt count =+ PSQL.query conn+ [sql| SELECT msg_id, read_ct, enqueued_at, NULL, vt, message FROM pgmq.read(?, ?, ?) |]+ (queue, vt, count)++{-| Reads a single message, polling for given duration if the queue+ is empty.+ + NOTE This is a blocking operation.+ + https://tembo.io/pgmq/api/sql/functions/#read_with_poll -}+readMessageWithPoll :: (SerializableMessage a)+ => PSQL.Connection+ -> Queue+ -> VisibilityTimeout+ -> MaxPollSeconds+ -> PollIntervalMs+ -> IO (Maybe (Message a))+readMessageWithPoll conn queue vt maxPollSeconds pollIntervalMs =+ readMessagesWithPoll conn queue vt 1 maxPollSeconds pollIntervalMs <&> headMay++-- | Reads given number of messages, polling for given duration if the+-- queue is empty.+-- NOTE This is a blocking operation.+-- https://tembo.io/pgmq/api/sql/functions/#read_with_poll+readMessagesWithPoll :: (SerializableMessage a)+ => PSQL.Connection+ -> Queue+ -> VisibilityTimeout+ -> MessageCount+ -> MaxPollSeconds+ -> PollIntervalMs+ -> IO [Message a]+readMessagesWithPoll conn queue vt count maxPollSeconds pollIntervalMs =+ PSQL.query conn+ [sql| SELECT msg_id, read_ct, enqueued_at, NULL, vt, message FROM pgmq.read_with_poll(?, ?, ?, ?, ?) |]+ (queue, vt, count, maxPollSeconds, pollIntervalMs)++-- | Sends one message to a queue+-- https://tembo.io/pgmq/api/sql/functions/#send+sendMessage :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> a -> Delay -> IO Int+sendMessage conn queue msg delay = do+ [PSQL.Only msgId] <- PSQL.query conn [sql| SELECT pgmq.send(?, ?::jsonb, ?) |] (queue, PSQL.Aeson msg, delay)++ return msgId+++-- | Sends a batch of messages+-- https://tembo.io/pgmq/api/sql/functions/#send_batch+sendMessages :: (SerializableMessage a)+ => PSQL.Connection -> Queue -> [a] -> Delay -> IO [Int]+sendMessages conn queue msgs delay = do+ ret <- PSQL.query conn [sql| SELECT pgmq.send_batch(?, ?::jsonb[], ?) |] (queue, PSQL.PGArray (PSQL.Aeson <$> msgs), delay)++ return $ (\(PSQL.Only id') -> id') <$> ret++-- | Sets the visibility timeout of a message for X seconds from now+-- https://tembo.io/pgmq/api/sql/functions/#set_vt+setMessageVt :: PSQL.Connection -> Queue -> MessageId -> VisibilityTimeout -> IO ()+setMessageVt conn queue msgId vt =+ void (PSQL.query conn [sql| SELECT 1 FROM pgmq.set_vt(?, ?, ?) |] (queue, msgId, vt) :: IO [PSQL.Only Int])++++-- | This function returns the number of readily available messages.+-- 'metrics' returns total queue size, but even then, when vt of+-- these messages is set in the future, the read operation might+-- return nothing (see queueAvailableNow).+queueAvailableLength :: PSQL.Connection -> Queue -> IO Int+queueAvailableLength conn queue = do+ let queueTable = "q_" <> queue+ let tableId = PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack queueTable+ [PSQL.Only len] <- PSQL.query conn [sql| SELECT COUNT(*) FROM ? WHERE vt <= now() |] (PSQL.Only tableId) :: IO [PSQL.Only Int]+ return len++-- | Same as 'queryAvailableLength' but returns the ids of messages.+queueAvailableIds :: PSQL.Connection -> Queue -> IO [MessageId]+queueAvailableIds conn queue = do+ let queueTable = "q_" <> queue+ let tableId = PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack queueTable+ ret <- PSQL.query conn [sql| SELECT msg_id FROM ? WHERE vt <= now() |] (PSQL.Only tableId) :: IO [PSQL.Only MessageId]+ return $ (\(PSQL.Only msgId) -> msgId) <$> ret+++{-|+ A utility function: sometimes pgmq throws an error that a table (for+ queue) doesn't exist and we want to ignore it as it's not critical in+ the given function.++ '42P01' means 'undefined_table' in postgres:+ https://www.postgresql.org/docs/current/errcodes-appendix.html+-}+-- withNoDoesNotExistError :: IO a -> IO a
+ src/Database/PGMQ/Types.hs view
@@ -0,0 +1,120 @@+{-|+Module : Database.PGMQ.Types+Description : PGMQ types+Copyright : (c) Gargantext, 2024-Present+License : AGPL+Maintainer : gargantext@iscpif.fr+Stability : experimental+Portability : POSIX+-}++{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+ +module Database.PGMQ.Types+where++import Data.Aeson (FromJSON, ToJSON)+import Data.Time.LocalTime (ZonedTime, zonedTimeToUTC)+import Data.Typeable (Typeable)+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.FromRow qualified as PSQL (field, FromRow(..))+import Database.PostgreSQL.Simple.Newtypes qualified as PSQL (getAeson)+++-- | Time in seconds before the message becomes visible (used when+-- | sending message to pgmq).+type Delay = Int -- in seconds+-- | Number of messages in metrics report+type MessageCount = Int+-- | The ID of a message in pgmq+type MessageId = Int+-- | Name of queue in pgmq+type Queue = String+-- | Number of times a message was read from pgmq+type ReadCount = Int+-- | https://tembo.io/pgmq/#visibility-timeout-vt+-- (in seconds)+-- pgmq guarantees exactly once delivery of a message within a+-- visibility timeout. The visibility timeout is the amount of time a+-- message is invisible to other consumers after it has been read by+-- a consumer. If the message is NOT deleted or archived within the+-- visibility timeout, it will become visible again and can be read+-- by another consumer.+type VisibilityTimeout = Int++-- | 'MaxPollSeconds' and 'PollIntervalMs' are used in 'pgmq.read_with_poll'+type MaxPollSeconds = Int+type PollIntervalMs = Int++ +-- | Basic message typeclass for PGMQ: it has to be jsonb-serializable+type SerializableMessage a = ( FromJSON a+ , ToJSON a+ -- NOTE This shouldn't be necessary+ , Typeable a )++-- | Message, as returned by the 'pgmq.read' function+data Message a =+ Message { msgId :: MessageId+ , readCt :: ReadCount+ , enqueuedAt :: ZonedTime+ , archivedAt :: Maybe ZonedTime+ , vt :: ZonedTime+ , message :: a }+ deriving (Show)+-- NOTE I'm not sure if this is needed+instance Eq a => Eq (Message a) where+ (==) msg1 msg2 =+ (msgId msg1) == (msgId msg2) &&+ (readCt msg1) == (readCt msg2) &&+ (zonedTimeToUTC $ vt msg1) == (zonedTimeToUTC $ vt msg2) &&+ (message msg1) == (message msg2)+instance SerializableMessage a => PSQL.FromRow (Message a) where+ fromRow = do+ msgId <- PSQL.field+ readCt <- PSQL.field+ enqueuedAt <- PSQL.field+ archivedAt <- PSQL.field+ vt <- PSQL.field+ messageA <- PSQL.field++ return $ Message { message = PSQL.getAeson messageA, .. }++-- | Returned by pgmq.list_queues+data QueueInfo =+ QueueInfo { queueName :: Queue+ , isPartitioned :: Bool+ , isUnlogged :: Bool+ , createdAt :: ZonedTime }+ deriving (Show)+instance PSQL.FromRow QueueInfo where+ fromRow = do+ queueName <- PSQL.field+ isPartitioned <- PSQL.field+ isUnlogged <- PSQL.field+ createdAt <- PSQL.field++ return $ QueueInfo { .. }+++-- | Returned by pgmq.metrics, pgmq.metrics_all+data Metrics =+ Metrics { queueName :: Queue+ , queueLength :: Int+ , newestMsgAgeSec :: Maybe Int+ , oldestMsgAgeSec :: Maybe Int+ , totalMessages :: MessageCount+ , scrapeTime :: ZonedTime }+ deriving (Show)+instance PSQL.FromRow Metrics where+ fromRow = do+ queueName <- PSQL.field+ queueLength <- PSQL.field+ newestMsgAgeSec <- PSQL.field+ oldestMsgAgeSec <- PSQL.field+ totalMessages <- PSQL.field+ scrapeTime <- PSQL.field++ return $ Metrics { .. }
+ tests/Test/Integration/PGMQ/Helpers.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE QuasiQuotes #-}++module Test.Integration.PGMQ.Helpers+ ( pgmqHelpersTests )+where++import Control.Exception (bracket)+import Data.Text qualified as T+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.Newtypes qualified as PSQL (Aeson(..))+import Database.PostgreSQL.Simple.SqlQQ (sql)+import Database.PostgreSQL.Simple.Types qualified as PSQL (QualifiedIdentifier(..))+import Database.PGMQ qualified as PGMQ+import Test.Hspec+import Test.Integration.Utils (getPSQLEnvConnectInfo, randomQueueName)+import Test.RandomStrings (randomASCII, randomString, onlyAlphaNum)+++data TestEnv =+ TestEnv {+ conn :: PSQL.Connection+ , queue :: PGMQ.Queue+ }+ +-- NOTE These tests expect a local pgmq server runnign on port 5432.++testQueuePrefix :: PGMQ.Queue+testQueuePrefix = "test_pgmq"++setUpConn :: IO TestEnv+setUpConn = do+ connInfo <- getPSQLEnvConnectInfo+ conn <- PSQL.connect connInfo+ queue <- randomQueueName testQueuePrefix+ return $ TestEnv { conn, queue }++dropConn :: TestEnv -> IO ()+dropConn (TestEnv { conn }) = do+ PSQL.close conn++withConn :: (TestEnv -> IO ()) -> IO ()+withConn = bracket setUpConn dropConn++withPGMQ :: (TestEnv -> IO ()) -> IO ()+withPGMQ f = withConn $ \testEnv -> bracket (setUpPGMQ testEnv) (tearDownPGMQ testEnv) (\_ -> f testEnv)+ where+ setUpPGMQ (TestEnv { conn, queue }) = do+ PGMQ.initialize conn+ PGMQ.createQueue conn queue++ tearDownPGMQ (TestEnv { conn = _conn, queue = _queue }) _ = do+ -- PGMQ.dropQueue conn queue+ pure ()++ +pgmqHelpersTests :: Spec+pgmqHelpersTests = parallel $ around withPGMQ $ describe "PGMQ Helpers" $ do+ it "can send message when queue doesn't exist" $ \(TestEnv { conn, queue }) -> do+ let newQueue = queue <> "_new"+ let queueId = PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack ("q_" <> newQueue)+ let iter = [1..20] :: [Int] -- number of steps+ mapM_ (\_i -> do+ -- Generate random strings and make sure that the+ -- message ids we get from sendMessage match our data+ message <- randomString (onlyAlphaNum randomASCII) 20+ msgId <- PGMQ.safeSendMessage conn newQueue message 0+ [PSQL.Only (PSQL.Aeson msg)] <- PSQL.query conn [sql| SELECT message FROM ? WHERE msg_id = ? |]+ (queueId, msgId)+ msg `shouldBe` message+ ) iter
+ tests/Test/Integration/PGMQ/Simple.hs view
@@ -0,0 +1,138 @@+{-# LANGUAGE QuasiQuotes #-}++module Test.Integration.PGMQ.Simple+ ( pgmqSimpleTests )+where++import Control.Exception (bracket)+-- import Data.Aeson qualified as Aeson+import Data.Maybe (isJust, isNothing, fromJust)+import Data.Text qualified as T+import Database.PostgreSQL.Simple qualified as PSQL+import Database.PostgreSQL.Simple.Newtypes qualified as PSQL (Aeson(..))+import Database.PostgreSQL.Simple.SqlQQ (sql)+import Database.PostgreSQL.Simple.Types qualified as PSQL (QualifiedIdentifier(..))+import Database.PGMQ.Simple qualified as PGMQ+import Database.PGMQ.Types qualified as PGMQ+import Test.Hspec+import Test.Integration.Utils (getPSQLEnvConnectInfo, randomQueueName)+import Test.RandomStrings (randomASCII, randomString, onlyAlphaNum)+++data TestEnv =+ TestEnv {+ conn :: PSQL.Connection+ , queue :: PGMQ.Queue+ }+ +-- NOTE These tests expect a local pgmq server runnign on port 5432.++testQueuePrefix :: PGMQ.Queue+testQueuePrefix = "test_pgmq"++setUpConn :: IO TestEnv+setUpConn = do+ connInfo <- getPSQLEnvConnectInfo+ conn <- PSQL.connect connInfo+ queue <- randomQueueName testQueuePrefix+ return $ TestEnv { conn, queue }++dropConn :: TestEnv -> IO ()+dropConn (TestEnv { conn }) = do+ PSQL.close conn++withConn :: (TestEnv -> IO ()) -> IO ()+withConn = bracket setUpConn dropConn++withPGMQ :: (TestEnv -> IO ()) -> IO ()+withPGMQ f = withConn $ \testEnv -> bracket (setUpPGMQ testEnv) (tearDownPGMQ testEnv) (\_ -> f testEnv)+ where+ setUpPGMQ (TestEnv { conn, queue }) = do+ PGMQ.initialize conn+ PGMQ.createQueue conn queue++ tearDownPGMQ (TestEnv { conn = _conn, queue = _queue }) _ = do+ -- PGMQ.dropQueue conn queue+ pure ()++ +pgmqSimpleTests :: Spec+pgmqSimpleTests = parallel $ around withPGMQ $ describe "PGMQ Simple" $ do+ -- it "can get metrics for non-existing queue" $ \(TestEnv { conn, queue }) -> do+ -- -- first of all, this should also work for non-existing queues+ -- metrics <- PGMQ.getMetrics conn queue+ -- metrics `shouldSatisfy` isNothing++ it "can get metrics for an empty queue" $ \(TestEnv { conn, queue }) -> do+ metrics <- PGMQ.getMetrics conn queue+ metrics `shouldSatisfy` isJust+ (PGMQ.queueLength <$> metrics) `shouldBe` Just 0++ it "listQueues properly returns our queue" $ \(TestEnv { conn, queue }) -> do+ queues <- PGMQ.listQueues conn+ ((\(PGMQ.QueueInfo { queueName }) -> queueName) <$> queues) `shouldContain` [queue]+ + it "can archive messages properly and read them" $ \(TestEnv { conn, queue }) -> do+ let message = "hello" :: String+ msgId <- PGMQ.sendMessage conn queue message 0+ mMsg <- PGMQ.readMessage conn queue 0 :: IO (Maybe (PGMQ.Message String))+ mMsg `shouldSatisfy` isJust+ let (PGMQ.Message { msgId = msgId', message = message', archivedAt }) = fromJust mMsg+ message `shouldBe` message'+ msgId `shouldBe` msgId'+ archivedAt `shouldSatisfy` isNothing+ PGMQ.archiveMessage conn queue msgId+ mMsgArchive <- PGMQ.readMessageFromArchive conn queue msgId+ mMsgArchive `shouldSatisfy` isJust+ let (PGMQ.Message { msgId = msgArchiveId, message = messageArchive, archivedAt = archivedAt' }) = fromJust mMsgArchive+ msgId `shouldBe` msgArchiveId+ message `shouldBe` messageArchive+ archivedAt' `shouldSatisfy` isJust++ it "send message properly returns msg id" $ \(TestEnv { conn, queue }) -> do+ let queueId = PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack ("q_" <> queue)+ let iter = [1..20] :: [Int] -- number of steps+ mapM_ (\_i -> do+ -- Generate random strings and make sure that the+ -- message ids we get from sendMessage match our data+ message <- randomString (onlyAlphaNum randomASCII) 20+ msgId <- PGMQ.sendMessage conn queue message 0+ [PSQL.Only (PSQL.Aeson msg)] <- PSQL.query conn [sql| SELECT message FROM ? WHERE msg_id = ? |]+ (queueId, msgId)+ msg `shouldBe` message+ ) iter++ it "send messages properly return msg ids" $ \(TestEnv { conn, queue }) -> do+ let queueId = PSQL.QualifiedIdentifier (Just "pgmq") $ T.pack ("q_" <> queue)+ let iter = [1..20] :: [Int] -- number of steps+ mapM_ (\_i -> do+ -- Generate random strings and make sure that the+ -- message ids we get from sendMessage match our data+ message1 <- randomString (onlyAlphaNum randomASCII) 20+ message2 <- randomString (onlyAlphaNum randomASCII) 20+ [msgId1, msgId2] <- PGMQ.sendMessages conn queue [message1, message2] 0+ [PSQL.Only (PSQL.Aeson msg1)] <- PSQL.query conn [sql| SELECT message FROM ? WHERE msg_id = ? |]+ (queueId, msgId1)+ [PSQL.Only (PSQL.Aeson msg2)] <- PSQL.query conn [sql| SELECT message FROM ? WHERE msg_id = ? |]+ (queueId, msgId2)+ msg1 `shouldBe` message1+ msg2 `shouldBe` message2+ ) iter++ it "can count messages properly, including vt (https://github.com/tembo-io/pgmq/issues/301)" $ \(TestEnv { conn, queue }) -> do+ let message = "hello vt test" :: String+ msgId <- PGMQ.sendMessage conn queue message 1+ -- immediately after creating such message, we should get metrics.length = 1 and available length = 0+ mMetrics <- PGMQ.getMetrics conn queue+ let mQLen = PGMQ.queueLength <$> mMetrics+ mQLen `shouldBe` (Just 1)+ aLen <- PGMQ.queueAvailableLength conn queue+ aLen `shouldBe` 0++ -- Now reset the vt+ PGMQ.setMessageVt conn queue msgId 0+ mMetrics' <- PGMQ.getMetrics conn queue+ let mQLen' = PGMQ.queueLength <$> mMetrics'+ mQLen' `shouldBe` (Just 1)+ aLen' <- PGMQ.queueAvailableLength conn queue+ aLen' `shouldBe` 1
+ tests/Test/Integration/Utils.hs view
@@ -0,0 +1,28 @@+module Test.Integration.Utils+ ( getPSQLEnvConnectInfo+ , randomQueueName )+where++import Data.Maybe (fromMaybe)+import Database.PGMQ.Types (Queue)+import Database.PostgreSQL.Simple qualified as PSQL+import System.Environment (lookupEnv)+import Test.RandomStrings (randomASCII, randomString, onlyLower)+++getPSQLEnvConnectInfo :: IO PSQL.ConnectInfo+getPSQLEnvConnectInfo = do+ pgUser <- lookupEnv "POSTGRES_USER"+ pgDb <- lookupEnv "POSTGRES_DB"+ pgPass <- lookupEnv "POSTGRES_PASSWORD"+ pgHost <- lookupEnv "POSTGRES_HOST"+ -- https://hackage.haskell.org/package/postgresql-simple-0.7.0.0/docs/Database-PostgreSQL-Simple.html#t:ConnectInfo+ pure $ PSQL.defaultConnectInfo { PSQL.connectUser = fromMaybe "postgres" pgUser+ , PSQL.connectDatabase = fromMaybe "postgres" pgDb+ , PSQL.connectHost = fromMaybe "localhost" pgHost+ , PSQL.connectPassword = fromMaybe "postgres" pgPass }++randomQueueName :: Queue -> IO Queue+randomQueueName prefix = do+ postfix <- randomString (onlyLower randomASCII) 10+ return $ prefix <> "_" <> postfix
+ tests/integration-tests.hs view
@@ -0,0 +1,20 @@+module Main where++import Test.Integration.PGMQ.Helpers (pgmqHelpersTests)+import Test.Integration.PGMQ.Simple (pgmqSimpleTests)+import Test.Tasty+import Test.Tasty.Hspec++++main :: IO ()+main = do+ pgmqHelpersSpec <- testSpec "pgmqHelpersTests" pgmqHelpersTests+ pgmqSimpleSpec <- testSpec "pgmqSimpleTests" pgmqSimpleTests++ defaultMain $ testGroup "integration tests"+ [+ pgmqSimpleSpec+ , pgmqHelpersSpec+ ]+